diff --git a/README.md b/README.md index 1f6b1655..0253ec55 100644 --- a/README.md +++ b/README.md @@ -918,6 +918,59 @@ You can also specify bootstrap peers explicitly: --- +## Health Endpoint + +The local health endpoint is **off by default**. Enable it with `--metrics-port 9100` +or `ANT_METRICS_PORT=9100`; `payment.metrics_port` is the configuration setting. +Set it to `0` and restart to disable it. It binds only to IPv4 loopback +(`127.0.0.1`), not the node's public address. If the port is busy, the node warns +and continues without the endpoint; failure to publish its port also does not +prevent the node from running. Logging must be enabled to see these warnings. + +- `/health` is JSON for people and programs: `curl http://127.0.0.1:9100/health`. +- `/metrics` is Prometheus text for monitoring systems scraping on the same machine. +- `{root_dir}/metrics.port` contains the bound port followed by a newline. It is + cleaned up best-effort on orderly shutdown, when disabled, or when binding fails. + A crash or filesystem error can leave it behind: treat it as discovery, not + proof, and verify the response's `peer_id` before trusting it. + +Both endpoints read the same existing node state when requested. `peer_id` (hex) +and `version` are descriptive JSON strings and labels on all Prometheus series. +The remaining JSON fields and matching metrics are: + +| JSON field | Prometheus metric | Meaning | +|-----------|-------------------|---------| +| `uptime_secs` | `ant_uptime_seconds` | Seconds since the P2P node was constructed | +| `bootstrapped` | `p2p_health_status` | Initial bootstrap completed, **not** a verdict that the node is healthy | +| `peer_count` | `p2p_network_peer_count` | Currently connected authenticated peers | +| `routing_table_size` | `p2p_dht_routing_table_size` | Current peer entries in the routing table | +| `storage_enabled` | `ant_storage_enabled` | Whether this node provides storage | +| `chunks_current` | `ant_chunks_current` | Chunks currently persisted, including existing chunks after restart | +| `chunks_written_total` | `ant_chunks_written_total` | Successful chunk writes since process start | +| `bytes_written_total` | `ant_bytes_written_total` | Bytes written since process start | +| `chunks_served_total` | `ant_chunks_served_total` | Storage reads since process start, not completed client downloads | +| `bytes_served_total` | `ant_bytes_served_total` | Bytes read from storage since process start | + +`_total` fields are process-lifetime activity counters and reset on restart; +other numeric and boolean fields report current state. Fields are additive. +JSON uses booleans for `bootstrapped` and `storage_enabled`; Prometheus uses +gauges with `0`/`1`. Other live measurements are gauges and `_total` metrics are +counters. Storage measurements are zero when storage is disabled. Activity +counters reflect the file store; they do not include legacy-store activity during +migration. No live stored-byte metric is emitted. + +The inherited `p2p_` and `ant_` names match existing local deployment tooling +where meanings agree. Remote Prometheus worker-IP targets cannot reach this +loopback listener, and dashboard panels for deferred metrics remain empty. +Requests are read-only, served one at a time with bounded headers and a timeout; +shutdown cancels active requests. A single `Host` header must name `127.0.0.1` or +`localhost`, optionally followed by the bound port. Missing, duplicate or other +host values return `403` without health data; this also rejects website hostnames +that resolve to loopback. Normal curl and Prometheus requests supply this header. +For trusted hosts, only `GET /health` and `GET /metrics` are supported; other paths +return `404`, other methods `405` with `Allow: GET`. Responses close the connection, +disable caching, and have no CORS header. + ## CLI Reference ``` diff --git a/config/production.toml b/config/production.toml index 72e82112..e77e337d 100644 --- a/config/production.toml +++ b/config/production.toml @@ -36,8 +36,8 @@ cache_capacity = 100000 # EVM network: "arbitrum-one" or "arbitrum-sepolia" evm_network = "arbitrum-one" -# Prometheus metrics port (0 to disable) -metrics_port = 9100 +# Loopback health JSON and Prometheus port (0 = off; set e.g. 9100 to enable) +metrics_port = 0 # --- Storage --- [storage] diff --git a/docs/adr/ADR-0017-local-node-health-endpoint.md b/docs/adr/ADR-0017-local-node-health-endpoint.md new file mode 100644 index 00000000..11822c15 --- /dev/null +++ b/docs/adr/ADR-0017-local-node-health-endpoint.md @@ -0,0 +1,82 @@ +# ADR-0017: Local node health endpoint + +- **Status:** Proposed +- **Date:** 2026-09-30 +- **Decision owners:** ant-node maintainers (review pending) +- **Reviewers:** Jim Collinson; ant-node maintainers (review pending) +- **Supersedes:** none +- **Superseded by:** none +- **Related:** [V2-1380](https://linear.app/autonominetwork/issue/V2-1380/ant-node-serve-a-local-health-endpoint-on-the-existing-metrics-port) + +## Context + +The node already holds information about its connectivity, storage and uptime, +but local tools cannot ask for it. Its configuration advertises a monitoring +endpoint that has not served anything. Operators and applications instead see +only process state or must inspect logs. Logging is optional, and its messages +are not a stable interface. + +Starting a listener on existing nodes could conflict with other local services. +Ordinary deployments should remain unchanged unless an operator enables it. + +## Decision Drivers + +- Make existing state available without requiring logging or new instrumentation. +- Support people, applications and standard monitoring tools. +- Keep the node private, dependency-light and unaffected by diagnostic failures. +- Require an explicit choice to enable the service. + +## Considered Options + +1. **Read logs:** rejected because messages change and logging may be disabled. +2. **Use a private channel through the management daemon:** rejected because it + couples the node to one consumer and requires a larger coordinated change. +3. **Let the node answer local, read-only health queries:** chosen because each + consumer can ask directly for the existing state. + +## Decision + +The node will offer its own health information on demand through a local endpoint. +We commit to these boundaries: + +- It binds only to loopback, meaning it is reachable only on the same machine. +- It is read-only and discloses no secrets, network addresses or filesystem paths. +- A form for programs and a form for monitoring carry the same values. +- Its failures cannot fail node startup, and its work cannot hold shutdown open + indefinitely. +- It advertises its bound port locally for tools to discover. +- It is off unless explicitly configured on; configuration changes apply at restart. + +The concrete interface and compatibility contract belong in the README and +linked issue's specification, not in this decision record. + +## Consequences + +### Positive + +- Local consumers obtain health without inspecting logs or storage files. +- Existing in-memory state is reused without adding dependencies. +- Nodes that do not enable the endpoint open no new listener. + +### Negative / Trade-offs + +- Any local process can read the published health when enabled. +- A busy port leaves that node without the endpoint rather than preventing startup. +- Remote monitoring needs a separate, deliberately configured access route. + +### Neutral / Operational + +- Discovery identifies where to ask, not proof that a node is healthy. +- The initial information set is intentionally limited; more instrumentation is + separate work. + +## Validation + +Tests and a development-network run check both representations, disabled mode, +port conflicts, local discovery and shutdown. Review confirms local-only binding, +absence of sensitive information, and isolation from node operation. + +## Notes for AI-assisted work + +AI tools may draft this record but must not mark it Accepted. Human engineering +review owns acceptance; Accepted records are immutable. diff --git a/src/bin/ant-node/cli.rs b/src/bin/ant-node/cli.rs index d5575d7c..8c7e11e1 100644 --- a/src/bin/ant-node/cli.rs +++ b/src/bin/ant-node/cli.rs @@ -98,9 +98,9 @@ pub struct Cli { #[arg(long, env = "ANT_EVM_PAYMENT_VAULT")] pub evm_payment_vault: Option, - /// Metrics port for Prometheus scraping (0 to disable). - #[arg(long, default_value = "9100", env = "ANT_METRICS_PORT")] - pub metrics_port: u16, + /// Loopback health JSON and Prometheus port (0 = off; set e.g. 9100 to enable). + #[arg(long, env = "ANT_METRICS_PORT")] + pub metrics_port: Option, /// Enable logging output. /// When omitted, the tracing subscriber is not installed and no log @@ -302,7 +302,7 @@ impl Cli { cache_capacity: self.cache_capacity, rewards_address: self.rewards_address, evm_network, - metrics_port: self.metrics_port, + metrics_port: self.metrics_port.unwrap_or(config.payment.metrics_port), }; // Determine bootstrap source and apply auto-discovery if needed. @@ -374,6 +374,99 @@ mod tests { use clap::Parser; use std::net::{IpAddr, Ipv4Addr}; + #[test] + fn metrics_port_default_is_off() -> Result<(), Box> { + use clap::CommandFactory; + let command = Cli::command().mut_arg("metrics_port", |arg| arg.env(None::<&str>)); + let matches = command.try_get_matches_from(["ant-node"])?; + let cli = ::from_arg_matches(&matches)?; + assert_eq!(cli.into_config()?.0.payment.metrics_port, 0); + Ok(()) + } + + #[test] + fn metrics_port_preserves_config_and_explicit_overrides( + ) -> Result<(), Box> { + use clap::{CommandFactory, FromArgMatches}; + let dir = tempfile::tempdir()?; + let path = dir.path().join("node.toml"); + std::fs::write(&path, "[payment]\nmetrics_port = 23456\n")?; + let path = path.to_str().ok_or("config path must be UTF-8")?; + for (extra, expected) in [ + (vec![], 23456), + (vec!["--metrics-port", "34567"], 34567), + (vec!["--metrics-port", "0"], 0), + ] { + let command = Cli::command().mut_arg("metrics_port", |arg| arg.env(None::<&str>)); + let args = [vec!["ant-node", "--config", path], extra].concat(); + let matches = command.try_get_matches_from(args)?; + let cli = Cli::from_arg_matches(&matches)?; + assert_eq!(cli.into_config()?.0.payment.metrics_port, expected); + } + Ok(()) + } + + #[test] + fn metrics_port_environment_and_cli_precedence() -> Result<(), Box> { + const CHILD_EXPECTED: &str = "ANT_TEST_METRICS_PORT_EXPECTED"; + const CHILD_CONFIG: &str = "ANT_TEST_METRICS_PORT_CONFIG"; + const CHILD_CLI: &str = "ANT_TEST_METRICS_PORT_CLI"; + + if let Ok(expected) = std::env::var(CHILD_EXPECTED) { + let mut args = vec![ + "ant-node".to_string(), + "--config".to_string(), + std::env::var(CHILD_CONFIG)?, + ]; + if let Ok(port) = std::env::var(CHILD_CLI) { + args.extend(["--metrics-port".to_string(), port]); + } + let cli = Cli::try_parse_from(args)?; + assert_eq!( + cli.into_config()?.0.payment.metrics_port, + expected.parse::()? + ); + return Ok(()); + } + + let dir = tempfile::tempdir()?; + let path = dir.path().join("node.toml"); + std::fs::write(&path, "[payment]\nmetrics_port = 23456\n")?; + for (env_port, cli_port, expected) in [ + (None, None, 23456), + (Some("34567"), None, 34567), + (Some("0"), None, 0), + (Some("0"), Some("45678"), 45678), + (Some("34567"), Some("0"), 0), + ] { + // Isolate clap's real environment bindings without changing this process's env. + let mut child = std::process::Command::new(std::env::current_exe()?); + child + .args([ + "--exact", + "cli::tests::metrics_port_environment_and_cli_precedence", + "--nocapture", + ]) + .env_clear() + .env(CHILD_EXPECTED, expected.to_string()) + .env(CHILD_CONFIG, &path); + if let Some(port) = env_port { + child.env("ANT_METRICS_PORT", port); + } + if let Some(port) = cli_port { + child.env(CHILD_CLI, port); + } + let output = child.output()?; + assert!( + output.status.success(), + "metrics env={env_port:?}, CLI={cli_port:?} failed:\n{}{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + } + Ok(()) + } + #[test] fn webrtc_direct_port_overrides_the_default_bind_port() -> Result<(), Box> { diff --git a/src/config.rs b/src/config.rs index 09b06144..0b851c95 100644 --- a/src/config.rs +++ b/src/config.rs @@ -382,8 +382,8 @@ pub struct PaymentConfig { #[serde(default)] pub evm_network: EvmNetworkConfig, - /// Metrics port for Prometheus scraping. - /// Set to 0 to disable metrics endpoint. + /// Loopback health JSON and Prometheus port. + /// 0 = off (default); set e.g. 9100 to enable. #[serde(default = "default_metrics_port")] pub metrics_port: u16, } @@ -400,7 +400,7 @@ impl Default for PaymentConfig { } const fn default_metrics_port() -> u16 { - 9100 + 0 } const fn default_cache_capacity() -> usize { @@ -780,6 +780,15 @@ mod tests { assert!(config.cache_capacity > 0, "Cache capacity must be positive"); } + #[test] + fn metrics_endpoint_defaults_to_off() { + assert_eq!(PaymentConfig::default().metrics_port, 0); + let absent: NodeConfig = toml::from_str("").unwrap(); + let partial: NodeConfig = toml::from_str("[payment]\ncache_capacity = 123").unwrap(); + assert_eq!(absent.payment.metrics_port, 0); + assert_eq!(partial.payment.metrics_port, 0); + } + #[test] fn default_webrtc_listener_tracks_compile_time_feature() { let config = WebRtcDirectConfig::default(); diff --git a/src/lib.rs b/src/lib.rs index 852c837a..8b8df781 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -56,6 +56,7 @@ pub mod node; pub mod payment; pub mod pointer; pub mod replication; +mod status; pub mod storage; pub mod upgrade; #[cfg(feature = "webrtc-direct")] diff --git a/src/node.rs b/src/node.rs index 56813244..6ef386c3 100644 --- a/src/node.rs +++ b/src/node.rs @@ -205,6 +205,7 @@ impl NodeBuilder { ant_protocol, replication_engine, protocol_task: None, + status_task: None, migration_task, protocol_children: TaskTracker::new(), #[cfg(feature = "webrtc-direct")] @@ -622,6 +623,8 @@ pub struct RunningNode { replication_engine: Option, /// Protocol message routing background task. protocol_task: Option>, + /// Sequential loopback health responder, cancelled and joined during cleanup. + status_task: Option>, /// The task moving this node off the legacy chunk store, if it has one. /// /// Awaited before the replication engine and the P2P layer are torn down, because it @@ -715,6 +718,15 @@ impl RunningNode { let listen_addrs = self.p2p_node.listen_addrs().await; info!(listen_addrs = ?listen_addrs, "P2P node started"); + self.status_task = crate::status::spawn( + self.config.payment.metrics_port, + &self.config.root_dir, + Arc::clone(&self.p2p_node), + self.ant_protocol.clone(), + self.shutdown.clone(), + ) + .await; + // Extract the actual bound port (config port may be 0 = auto-select) let actual_port = listen_addrs .first() @@ -955,6 +967,12 @@ impl RunningNode { /// Drain dependent work before shutting down native networking. async fn cleanup(&mut self) { self.shutdown.cancel(); + if let Some(task) = self.status_task.take() { + if let Err(error) = task.await { + warn!("Health endpoint task shutdown failed: {error}"); + } + } + crate::status::clear_port_file(&self.config.root_dir).await; // The shared token closes the WebRtcDirect accept loop and active // browser sessions before storage and native P2P are torn down. #[cfg(feature = "webrtc-direct")] @@ -1343,6 +1361,266 @@ mod tests { } } + async fn health_test_node(root: &Path, metrics_port: u16, storage: bool) -> RunningNode { + let mut config = local_node_config(root, rand::thread_rng().gen_range(TEST_PORT_RANGE)); + config.payment.metrics_port = metrics_port; + config.storage.enabled = storage; + config.webrtc_direct.enabled = false; + let mut node = NodeBuilder::new(config) + .build() + .await + .expect("build health test node"); + node.upgrade_monitor = None; + node + } + + async fn health_request(port: u16, request: &str) -> String { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + tokio::time::timeout(std::time::Duration::from_secs(5), async { + let mut stream = tokio::net::TcpStream::connect((std::net::Ipv4Addr::LOCALHOST, port)) + .await + .unwrap(); + stream.write_all(request.as_bytes()).await.unwrap(); + let mut response = String::new(); + stream.read_to_string(&mut response).await.unwrap(); + let (headers, body) = response.split_once("\r\n\r\n").unwrap(); + assert!(headers.contains("Connection: close\r\n")); + assert!(headers.contains("Cache-Control: no-store")); + assert!(!headers.contains("Access-Control")); + assert!(headers.contains(&format!("Content-Length: {}\r\n", body.len()))); + response + }) + .await + .expect("health request timed out") + } + + #[tokio::test] + #[allow(clippy::too_many_lines)] // One lifecycle: serving, collision, cancellation, cleanup. + async fn health_endpoints_run_collision_and_cleanup() { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + let dir = TempDir::new().unwrap(); + let root = dir.path().join("first"); + let reservation = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = reservation.local_addr().unwrap().port(); + drop(reservation); + let mut node = health_test_node(&root, port, true).await; + let peer_id = node.p2p_node.peer_id().to_hex(); + let shutdown = node.shutdown.clone(); + let mut events = node.subscribe_events(); + let task = tokio::spawn(async move { + node.run().await.expect("normal run"); + node + }); + assert!(matches!( + tokio::time::timeout(std::time::Duration::from_secs(10), events.recv()) + .await + .unwrap() + .unwrap(), + NodeEvent::Started + )); + assert_eq!( + std::fs::read_to_string(root.join("metrics.port")).unwrap(), + format!("{port}\n") + ); + let json = health_request(port, "GET /health HTTP/1.1\r\nHost: localhost\r\n\r\n").await; + assert!(json.starts_with("HTTP/1.1 200 OK\r\nContent-Type: application/json\r\n")); + let value: serde_json::Value = + serde_json::from_str(json.split_once("\r\n\r\n").unwrap().1).unwrap(); + assert_eq!(value["peer_id"], peer_id); + assert_eq!(value["version"], env!("CARGO_PKG_VERSION")); + assert_eq!(value["storage_enabled"], true); + assert!(value["bootstrapped"].is_boolean()); + assert_eq!(value.as_object().unwrap().len(), 12); + for field in [ + "uptime_secs", + "peer_count", + "routing_table_size", + "chunks_current", + "chunks_written_total", + "bytes_written_total", + "chunks_served_total", + "bytes_served_total", + ] { + assert!(value[field].is_u64(), "{field} must be numeric"); + } + let metrics = health_request( + port, + &format!("GET /metrics HTTP/1.1\r\nHost: 127.0.0.1:{port}\r\n\r\n"), + ) + .await; + assert!( + metrics.starts_with("HTTP/1.1 200 OK\r\nContent-Type: text/plain; version=0.0.4\r\n") + ); + assert!(metrics.contains("# TYPE ant_chunks_current gauge\n")); + assert!(metrics.contains("# TYPE ant_chunks_written_total counter\n")); + assert!(metrics.contains(&format!( + "peer_id=\"{peer_id}\",version=\"{}\"", + env!("CARGO_PKG_VERSION") + ))); + for (request, expected) in [ + ( + "GET /missing HTTP/1.1\r\nHost: localhost\r\n\r\n", + "404 Not Found", + ), + ( + "GET /health/ HTTP/1.1\r\nHost: localhost\r\n\r\n", + "404 Not Found", + ), + ( + "POST /health HTTP/1.1\r\nHost: localhost\r\n\r\n", + "405 Method Not Allowed", + ), + ( + "HEAD /metrics HTTP/1.1\r\nHost: localhost\r\n\r\n", + "405 Method Not Allowed", + ), + ] { + let response = health_request(port, request).await; + assert!(response.starts_with(&format!("HTTP/1.1 {expected}\r\n"))); + if expected == "405 Method Not Allowed" { + assert!(response.contains("\r\nAllow: GET\r\n")); + } + } + for path in ["/health", "/metrics"] { + for headers in [ + "", + "Host: example.com\r\n", + "Host: localhost.example.com\r\n", + "Host: localhost\r\nHost: example.com\r\n", + "Host: localhost:1\r\n", + ] { + let response = + health_request(port, &format!("GET {path} HTTP/1.1\r\n{headers}\r\n")).await; + assert!(response.starts_with("HTTP/1.1 403 Forbidden\r\n")); + assert_eq!( + response.split_once("\r\n\r\n").unwrap().1, + "{\"error\":\"untrusted host\"}" + ); + } + } + + // A second real node must start despite the first owning its configured port. + let second_root = dir.path().join("second"); + let mut second = health_test_node(&second_root, port, false).await; + std::fs::write(second_root.join("metrics.port"), "stale\n").unwrap(); + let second_shutdown = second.shutdown.clone(); + let mut second_events = second.subscribe_events(); + let second_task = tokio::spawn(async move { + second + .run() + .await + .expect("collision must not fail node startup"); + second + }); + assert!(matches!( + tokio::time::timeout(std::time::Duration::from_secs(10), second_events.recv()) + .await + .unwrap() + .unwrap(), + NodeEvent::Started + )); + assert!(!second_root.join("metrics.port").exists()); + second_shutdown.cancel(); + let second = tokio::time::timeout(std::time::Duration::from_secs(10), second_task) + .await + .unwrap() + .unwrap(); + assert!(second.status_task.is_none()); + + // Complete fragmented headers work; shutdown interrupts an incomplete one. + let mut stream = tokio::net::TcpStream::connect((std::net::Ipv4Addr::LOCALHOST, port)) + .await + .unwrap(); + stream.write_all(b"GET /hea").await.unwrap(); + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + stream + .write_all(b"lth HTTP/1.1\r\nHost: localhost\r\n\r\n") + .await + .unwrap(); + let mut response = String::new(); + tokio::time::timeout( + std::time::Duration::from_secs(5), + stream.read_to_string(&mut response), + ) + .await + .unwrap() + .unwrap(); + assert!(response.starts_with("HTTP/1.1 200 OK")); + let mut slow = tokio::net::TcpStream::connect((std::net::Ipv4Addr::LOCALHOST, port)) + .await + .unwrap(); + slow.write_all(b"GET /health HTTP/1.1\r\n").await.unwrap(); + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + shutdown.cancel(); + let mut byte = [0]; + assert_eq!( + tokio::time::timeout(std::time::Duration::from_secs(1), slow.read(&mut byte)) + .await + .unwrap() + .unwrap(), + 0 + ); + let node = tokio::time::timeout(std::time::Duration::from_secs(10), task) + .await + .unwrap() + .unwrap(); + assert!(node.status_task.is_none(), "responder must be joined"); + assert!(!root.join("metrics.port").exists()); + let _released = tokio::net::TcpListener::bind((std::net::Ipv4Addr::LOCALHOST, port)) + .await + .expect("cleanup releases socket"); + } + + #[tokio::test] + async fn health_disabled_clears_stale_file_and_storage_stats_are_zero() { + let dir = TempDir::new().unwrap(); + let reservation = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let enabled_port = reservation.local_addr().unwrap().port(); + drop(reservation); + for port in [0, enabled_port] { + let mut node = health_test_node(dir.path(), port, false).await; + std::fs::write(dir.path().join("metrics.port"), "stale\n").unwrap(); + let shutdown = node.shutdown.clone(); + let mut events = node.subscribe_events(); + let task = tokio::spawn(async move { + node.run().await.unwrap(); + node + }); + assert!(matches!( + tokio::time::timeout(std::time::Duration::from_secs(10), events.recv()) + .await + .unwrap() + .unwrap(), + NodeEvent::Started + )); + if port == 0 { + assert!(!dir.path().join("metrics.port").exists()); + } else { + let response = + health_request(port, "GET /health HTTP/1.1\r\nHost: localhost\r\n\r\n").await; + let value: serde_json::Value = + serde_json::from_str(response.split_once("\r\n\r\n").unwrap().1).unwrap(); + assert_eq!(value["storage_enabled"], false); + for field in [ + "chunks_current", + "chunks_written_total", + "bytes_written_total", + "chunks_served_total", + "bytes_served_total", + ] { + assert_eq!(value[field], 0); + } + } + shutdown.cancel(); + let node = tokio::time::timeout(std::time::Duration::from_secs(10), task) + .await + .unwrap() + .unwrap(); + assert!(node.status_task.is_none()); + assert!(!dir.path().join("metrics.port").exists()); + } + } + #[cfg(feature = "webrtc-direct")] #[tokio::test] async fn browser_startup_failure_drains_existing_migration() { diff --git a/src/status.rs b/src/status.rs new file mode 100644 index 00000000..849ec66b --- /dev/null +++ b/src/status.rs @@ -0,0 +1,338 @@ +//! Private, read-only loopback health responder. + +use crate::logging::warn; +use crate::storage::AntProtocol; +use saorsa_core::P2PNode; +use serde::Serialize; +use std::io; +use std::net::Ipv4Addr; +use std::path::Path; +use std::sync::Arc; +use std::time::Duration; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::{TcpListener, TcpStream}; +use tokio::task::JoinHandle; +use tokio_util::sync::CancellationToken; + +const REQUEST_TIMEOUT: Duration = Duration::from_secs(5); +const MAX_HEADER_BYTES: usize = 2048; + +#[derive(Serialize)] +struct Snapshot { + peer_id: String, + version: &'static str, + uptime_secs: u64, + bootstrapped: bool, + peer_count: usize, + routing_table_size: usize, + storage_enabled: bool, + chunks_current: u64, + chunks_written_total: u64, + bytes_written_total: u64, + chunks_served_total: u64, + bytes_served_total: u64, +} + +impl Snapshot { + async fn read(p2p: &P2PNode, protocol: Option<&AntProtocol>) -> Self { + let stats = protocol.map(AntProtocol::storage_stats).unwrap_or_default(); + Self { + peer_id: p2p.peer_id().to_hex(), + version: env!("CARGO_PKG_VERSION"), + uptime_secs: p2p.uptime().as_secs(), + bootstrapped: p2p.is_bootstrapped(), + peer_count: p2p.peer_count().await, + routing_table_size: p2p.dht_manager().get_routing_table_size().await, + storage_enabled: protocol.is_some(), + chunks_current: stats.current_chunks, + chunks_written_total: stats.chunks_stored, + bytes_written_total: stats.bytes_stored, + chunks_served_total: stats.chunks_retrieved, + bytes_served_total: stats.bytes_retrieved, + } + } + + fn metrics(&self) -> String { + // Both labels have controlled values: a hex peer ID and the package version. + let labels = format!("peer_id=\"{}\",version=\"{}\"", self.peer_id, self.version); + let mut text = String::new(); + for (name, kind, value) in [ + ("ant_uptime_seconds", "gauge", self.uptime_secs), + ("p2p_health_status", "gauge", u64::from(self.bootstrapped)), + ("p2p_network_peer_count", "gauge", self.peer_count as u64), + ( + "p2p_dht_routing_table_size", + "gauge", + self.routing_table_size as u64, + ), + ( + "ant_storage_enabled", + "gauge", + u64::from(self.storage_enabled), + ), + ("ant_chunks_current", "gauge", self.chunks_current), + ( + "ant_chunks_written_total", + "counter", + self.chunks_written_total, + ), + ( + "ant_bytes_written_total", + "counter", + self.bytes_written_total, + ), + ( + "ant_chunks_served_total", + "counter", + self.chunks_served_total, + ), + ("ant_bytes_served_total", "counter", self.bytes_served_total), + ] { + use std::fmt::Write; + let _ = writeln!(text, "# TYPE {name} {kind}\n{name}{{{labels}}} {value}"); + } + text + } +} + +/// Remove discovery left by this run or an earlier interrupted run. +pub async fn clear_port_file(root: &Path) { + match tokio::time::timeout( + REQUEST_TIMEOUT, + tokio::fs::remove_file(root.join("metrics.port")), + ) + .await + { + Ok(Ok(())) => {} + Ok(Err(error)) if error.kind() == io::ErrorKind::NotFound => {} + Ok(Err(error)) => warn!("Failed to clear health endpoint port file: {error}"), + Err(_) => warn!("Timed out clearing health endpoint port file"), + } +} + +/// Bind and publish best-effort; the caller owns cancellation and joining. +pub async fn spawn( + port: u16, + root: &Path, + p2p: Arc, + protocol: Option>, + shutdown: CancellationToken, +) -> Option> { + clear_port_file(root).await; + if port == 0 { + return None; + } + let listener = match TcpListener::bind((Ipv4Addr::LOCALHOST, port)).await { + Ok(listener) => listener, + Err(error) => { + warn!("Failed to bind health endpoint: {error}"); + return None; + } + }; + let address = match listener.local_addr() { + Ok(address) => address, + Err(error) => { + warn!("Failed to read health endpoint bound port: {error}"); + return None; + } + }; + match tokio::time::timeout( + REQUEST_TIMEOUT, + tokio::fs::write(root.join("metrics.port"), format!("{}\n", address.port())), + ) + .await + { + Ok(Ok(())) => {} + Ok(Err(error)) => warn!("Failed to publish health endpoint port: {error}"), + Err(_) => warn!("Timed out publishing health endpoint port"), + } + Some(tokio::spawn(async move { + loop { + let accepted = tokio::select! { + biased; + () = shutdown.cancelled() => break, + accepted = listener.accept() => accepted, + }; + let (mut stream, _) = match accepted { + Ok(accepted) => accepted, + Err(error) => { + warn!("Health endpoint accept failed: {error}"); + break; + } + }; + // No child tasks: cancellation drops even a fragmented read or blocked write. + tokio::select! { + biased; + () = shutdown.cancelled() => break, + _ = tokio::time::timeout(REQUEST_TIMEOUT, answer(&mut stream, &p2p, protocol.as_deref())) => {} + } + } + })) +} + +async fn answer( + stream: &mut TcpStream, + p2p: &P2PNode, + protocol: Option<&AntProtocol>, +) -> io::Result<()> { + let mut header = [0; MAX_HEADER_BYTES]; + let mut length = 0; + loop { + let n = stream.read(&mut header[length..]).await?; + if n == 0 { + return Ok(()); + } + length += n; + if let Some(end) = header[..length] + .windows(4) + .position(|bytes| bytes == b"\r\n\r\n") + { + length = end + 4; + break; + } + if length == header.len() { + return Ok(()); + } + } + let request = String::from_utf8_lossy(&header[..length]); + let mut tokens = request.lines().next().unwrap_or("").split_whitespace(); + let method = tokens.next().unwrap_or(""); + let path = tokens.next().unwrap_or(""); + let (status, content_type, body) = if !trusted_host(&request, stream.local_addr()?.port()) { + ( + "403 Forbidden", + "application/json", + "{\"error\":\"untrusted host\"}".to_string(), + ) + } else if method != "GET" { + ( + "405 Method Not Allowed", + "application/json", + "{\"error\":\"method not allowed\"}".to_string(), + ) + } else if path == "/health" || path == "/metrics" { + let snapshot = Snapshot::read(p2p, protocol).await; + if path == "/health" { + ( + "200 OK", + "application/json", + serde_json::to_string(&snapshot).map_err(io::Error::other)?, + ) + } else { + ("200 OK", "text/plain; version=0.0.4", snapshot.metrics()) + } + } else { + ( + "404 Not Found", + "application/json", + "{\"error\":\"not found\"}".to_string(), + ) + }; + let allow = if status == "405 Method Not Allowed" { + "Allow: GET\r\n" + } else { + "" + }; + let response = format!( + "HTTP/1.1 {status}\r\nContent-Type: {content_type}\r\nContent-Length: {}\r\n{allow}Connection: close\r\nCache-Control: no-store\r\n\r\n{body}", + body.len() + ); + stream.write_all(response.as_bytes()).await +} + +fn trusted_host(request: &str, port: u16) -> bool { + // Loopback binding alone does not reject a website whose DNS resolves locally. + let mut hosts = request + .lines() + .skip(1) + .take_while(|line| !line.is_empty()) + .filter_map(|line| { + let (name, value) = line.split_once(':')?; + name.eq_ignore_ascii_case("Host").then_some(value.trim()) + }); + let Some(host) = hosts.next() else { + return false; + }; + if hosts.next().is_some() { + return false; + } + let suffix = format!(":{port}"); + let host = host.strip_suffix(&suffix).unwrap_or(host); + host == "127.0.0.1" || host.eq_ignore_ascii_case("localhost") +} + +#[cfg(test)] +#[allow(clippy::unwrap_used)] +mod tests { + use super::*; + + #[test] + fn health_host_is_local_and_unambiguous() { + for host in [ + "localhost", + "LOCALHOST:23456", + "127.0.0.1", + "127.0.0.1:23456", + ] { + assert!(trusted_host( + &format!("GET /health HTTP/1.1\r\nhOsT: {host}\r\n\r\n"), + 23456 + )); + } + for host in [ + "", + "example.com", + "localhost.example.com", + "127.0.0.1.example.com", + "localhost:34567", + "127.0.0.1:34567", + "localhost,example.com", + "example.com@localhost", + ] { + assert!(!trusted_host( + &format!("GET /health HTTP/1.1\r\nHost: {host}\r\n\r\n"), + 23456 + )); + } + for request in [ + "GET /health HTTP/1.1\r\n\r\n", + "GET /health HTTP/1.1\r\nHost: localhost\r\nHost: localhost\r\n\r\n", + "GET /health HTTP/1.1\r\nHost: localhost\r\nhost: example.com\r\n\r\n", + "GET /health HTTP/1.1\r\n\r\nHost: localhost\r\n", + ] { + assert!(!trusted_host(request, 23456)); + } + } + + #[test] + fn health_json_and_metrics_golden() { + let snapshot = Snapshot { + peer_id: "abcd".into(), + version: "1.2.3", + uptime_secs: 12, + bootstrapped: true, + peer_count: 3, + routing_table_size: 4, + storage_enabled: true, + chunks_current: 5, + chunks_written_total: 6, + bytes_written_total: 7, + chunks_served_total: 8, + bytes_served_total: 9, + }; + assert_eq!(serde_json::to_string(&snapshot).unwrap(), + "{\"peer_id\":\"abcd\",\"version\":\"1.2.3\",\"uptime_secs\":12,\"bootstrapped\":true,\"peer_count\":3,\"routing_table_size\":4,\"storage_enabled\":true,\"chunks_current\":5,\"chunks_written_total\":6,\"bytes_written_total\":7,\"chunks_served_total\":8,\"bytes_served_total\":9}"); + assert_eq!(snapshot.metrics(), concat!( + "# TYPE ant_uptime_seconds gauge\nant_uptime_seconds{peer_id=\"abcd\",version=\"1.2.3\"} 12\n", + "# TYPE p2p_health_status gauge\np2p_health_status{peer_id=\"abcd\",version=\"1.2.3\"} 1\n", + "# TYPE p2p_network_peer_count gauge\np2p_network_peer_count{peer_id=\"abcd\",version=\"1.2.3\"} 3\n", + "# TYPE p2p_dht_routing_table_size gauge\np2p_dht_routing_table_size{peer_id=\"abcd\",version=\"1.2.3\"} 4\n", + "# TYPE ant_storage_enabled gauge\nant_storage_enabled{peer_id=\"abcd\",version=\"1.2.3\"} 1\n", + "# TYPE ant_chunks_current gauge\nant_chunks_current{peer_id=\"abcd\",version=\"1.2.3\"} 5\n", + "# TYPE ant_chunks_written_total counter\nant_chunks_written_total{peer_id=\"abcd\",version=\"1.2.3\"} 6\n", + "# TYPE ant_bytes_written_total counter\nant_bytes_written_total{peer_id=\"abcd\",version=\"1.2.3\"} 7\n", + "# TYPE ant_chunks_served_total counter\nant_chunks_served_total{peer_id=\"abcd\",version=\"1.2.3\"} 8\n", + "# TYPE ant_bytes_served_total counter\nant_bytes_served_total{peer_id=\"abcd\",version=\"1.2.3\"} 9\n", + )); + } +}