From 8f020faee4763f7846a105a37652fc0dfa0a1d78 Mon Sep 17 00:00:00 2001 From: LowderPlay Date: Mon, 24 Aug 2026 23:29:13 +0500 Subject: [PATCH] feat: node stats (#85) * feat: node stats * fix: remove ip --- docker-compose.yaml | 1 + nginx.conf | 5 +- ...0260824000000_reporter_connection_info.sql | 2 + website/src/api.rs | 3 + website/src/api/node_stats.rs | 192 ++++++++++++++++++ website/src/main.rs | 1 + website/src/mqtt.rs | 121 +++++++++-- website/src/mqtt_auth.rs | 25 ++- 8 files changed, 322 insertions(+), 28 deletions(-) create mode 100644 website/migrations/20260824000000_reporter_connection_info.sql create mode 100644 website/src/api/node_stats.rs diff --git a/docker-compose.yaml b/docker-compose.yaml index 1d3eb61..e234214 100644 --- a/docker-compose.yaml +++ b/docker-compose.yaml @@ -32,6 +32,7 @@ services: DATABASE_CACHE_DIR: /var/cache/cheburcheck/databases DATABASE_RETRY_INTERVAL_SECONDS: "${DATABASE_RETRY_INTERVAL_SECONDS:-300}" MQTT_ADMIN_TOKEN: "${MQTT_ADMIN_TOKEN}" + NODE_STATS_KEY: "${NODE_STATS_KEY}" MQTT_HOST: rmqtt MQTT_PORT: 11883 PROBE_TRACEROUTE_ENABLED: "${PROBE_TRACEROUTE_ENABLED:-false}" diff --git a/nginx.conf b/nginx.conf index feb0f4c..7b5b5b8 100644 --- a/nginx.conf +++ b/nginx.conf @@ -19,7 +19,7 @@ server { listen 80; server_name _; - set_real_ip_from 172.18.0.0/16; + set_real_ip_from 172.16.0.0/12; real_ip_header X-Forwarded-For; real_ip_recursive on; @@ -58,6 +58,9 @@ server { } location / { + proxy_buffer_size 32k; + proxy_buffers 8 32k; + proxy_busy_buffers_size 64k; proxy_pass http://frontend; } } diff --git a/website/migrations/20260824000000_reporter_connection_info.sql b/website/migrations/20260824000000_reporter_connection_info.sql new file mode 100644 index 0000000..35a20c3 --- /dev/null +++ b/website/migrations/20260824000000_reporter_connection_info.sql @@ -0,0 +1,2 @@ +ALTER TABLE reporters + ADD COLUMN IF NOT EXISTS last_connected_at TIMESTAMPTZ; diff --git a/website/src/api.rs b/website/src/api.rs index 4ae66ef..8c9908d 100644 --- a/website/src/api.rs +++ b/website/src/api.rs @@ -2,12 +2,15 @@ mod check_endpoint; #[path = "api/feedback.rs"] mod feedback_endpoint; +#[path = "api/node_stats.rs"] +mod nodes_endpoint; mod probe; mod rate_limit; mod status; pub use check_endpoint::check; pub use feedback_endpoint::feedback; +pub use nodes_endpoint::node_stats; pub use probe::probe_query; pub use rate_limit::{build_probe_rate_limiter, build_rate_limiter}; pub use status::{get_system_status, healthcheck}; diff --git a/website/src/api/node_stats.rs b/website/src/api/node_stats.rs new file mode 100644 index 0000000..a538370 --- /dev/null +++ b/website/src/api/node_stats.rs @@ -0,0 +1,192 @@ +use crate::mqtt::{MqttPublisher, ProbeStatusSnapshot}; +use rocket::http::Status; +use rocket::request::{FromRequest, Outcome}; +use rocket::serde::json::Json; +use rocket::{Request, State}; +use serde::Serialize; +use sqlx::PgPool; +use sqlx::types::chrono::{DateTime, Utc}; +use std::collections::HashMap; + +#[derive(sqlx::FromRow)] +struct ProbeMetadata { + id: i32, + name: String, + region: Option, + provider: Option, + asn: Option, + last_connected_at: Option>, +} + +pub struct NodeStatsKey; + +#[rocket::async_trait] +impl<'r> FromRequest<'r> for NodeStatsKey { + type Error = (); + + async fn from_request(request: &'r Request<'_>) -> Outcome { + let expected_key = std::env::var("NODE_STATS_KEY") + .ok() + .filter(|key| !key.is_empty()); + let provided_key = request + .headers() + .get_one("Authorization") + .and_then(|header| header.strip_prefix("Bearer ")); + + if valid_key(provided_key, expected_key.as_deref()) { + Outcome::Success(Self) + } else { + Outcome::Error((Status::Unauthorized, ())) + } + } +} + +#[derive(Debug, Serialize, PartialEq, Eq)] +pub struct NodeStatsResponse { + nodes: Vec, +} + +#[derive(Debug, Serialize, PartialEq, Eq)] +struct NodeStatus { + probe_id: i32, + name: String, + region: Option, + provider: Option, + asn: Option, + connected_at: Option>, + online: bool, + version: Option, + dpi_hop_v4: Option, + dpi_hop_v6: Option, +} + +#[get("/nodes")] +pub async fn node_stats( + _key: NodeStatsKey, + pool: &State, + mqtt: &State, +) -> Result, Status> { + let probes = sqlx::query_as::<_, ProbeMetadata>( + "SELECT id, name, region, provider, asn, last_connected_at + FROM reporters + ORDER BY id", + ) + .fetch_all(&**pool) + .await + .map_err(|error| { + log::error!("failed to load probe metadata for node stats: {error}"); + Status::InternalServerError + })?; + + let statuses = mqtt.probe_statuses().await; + Ok(Json(NodeStatsResponse { + nodes: build_node_statuses(probes, &statuses), + })) +} + +fn valid_key(provided: Option<&str>, expected: Option<&str>) -> bool { + provided + .zip(expected) + .is_some_and(|(provided, expected)| provided == expected && !provided.is_empty()) +} + +fn build_node_statuses( + probes: Vec, + statuses: &HashMap, +) -> Vec { + probes + .into_iter() + .map(|probe| { + let status = statuses.get(&probe.id.to_string()); + NodeStatus { + probe_id: probe.id, + name: probe.name, + region: probe.region, + provider: probe.provider, + asn: probe.asn, + connected_at: probe.last_connected_at, + online: status.is_some_and(|status| status.online), + version: status.map(|status| status.version.clone()), + dpi_hop_v4: status.and_then(|status| status.dpi_hop_v4), + dpi_hop_v6: status.and_then(|status| status.dpi_hop_v6), + } + }) + .collect() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn requires_a_configured_matching_key() { + assert!(valid_key(Some("secret"), Some("secret"))); + assert!(!valid_key(Some("wrong"), Some("secret"))); + assert!(!valid_key(None, Some("secret"))); + assert!(!valid_key(Some("secret"), None)); + assert!(!valid_key(Some(""), Some(""))); + } + + #[test] + fn builds_online_and_offline_json_nodes() { + let probes = vec![ + ProbeMetadata { + id: 1, + name: "Home router".to_string(), + region: Some("Ural".to_string()), + provider: Some("ExampleNet".to_string()), + asn: Some("AS64500".to_string()), + last_connected_at: Some("2026-08-24T12:34:56Z".parse().unwrap()), + }, + ProbeMetadata { + id: 2, + name: "Offline".to_string(), + region: None, + provider: None, + asn: None, + last_connected_at: Some("2026-08-23T12:34:56Z".parse().unwrap()), + }, + ]; + let statuses = HashMap::from([( + "1".to_string(), + ProbeStatusSnapshot { + online: true, + version: "1.2.3".to_string(), + dpi_hop_v4: Some(5), + dpi_hop_v6: None, + }, + )]); + + let nodes = build_node_statuses(probes, &statuses); + + assert_eq!( + nodes, + vec![ + NodeStatus { + probe_id: 1, + name: "Home router".to_string(), + region: Some("Ural".to_string()), + provider: Some("ExampleNet".to_string()), + asn: Some("AS64500".to_string()), + connected_at: Some("2026-08-24T12:34:56Z".parse().unwrap()), + online: true, + version: Some("1.2.3".to_string()), + dpi_hop_v4: Some(5), + dpi_hop_v6: None, + }, + NodeStatus { + probe_id: 2, + name: "Offline".to_string(), + region: None, + provider: None, + asn: None, + connected_at: Some("2026-08-23T12:34:56Z".parse().unwrap()), + online: false, + version: None, + dpi_hop_v4: None, + dpi_hop_v6: None, + }, + ] + ); + } +} diff --git a/website/src/main.rs b/website/src/main.rs index 796c6e8..c56d593 100644 --- a/website/src/main.rs +++ b/website/src/main.rs @@ -100,6 +100,7 @@ async fn rocket() -> _ { api::healthcheck, api::feedback, api::get_system_status, + api::node_stats, whitelist::histogram ], ) diff --git a/website/src/mqtt.rs b/website/src/mqtt.rs index c38ec52..421bb47 100644 --- a/website/src/mqtt.rs +++ b/website/src/mqtt.rs @@ -8,7 +8,7 @@ use rumqttc::{AsyncClient, Event as MqttEvent, Incoming, MqttOptions, QoS}; use serde::Deserialize; use sqlx::types::Uuid; use sqlx::types::chrono::Utc; -use std::collections::{HashMap, HashSet}; +use std::collections::HashMap; use std::fmt; use std::net::IpAddr; use std::sync::Arc; @@ -53,13 +53,22 @@ impl fmt::Display for PublishError { pub struct MqttPublisher { client: Option, sessions: Arc>>, - online_probes: Arc>>, + probe_statuses: ProbeStatuses, probe_config: Arc, task_timeout_ms: u64, } type ProbeResultSender = rocket::tokio::sync::broadcast::Sender; pub type ProbeResultReceiver = rocket::tokio::sync::broadcast::Receiver; +type ProbeStatuses = Arc>>; + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ProbeStatusSnapshot { + pub online: bool, + pub version: String, + pub dpi_hop_v4: Option, + pub dpi_hop_v6: Option, +} #[derive(Deserialize)] struct ProbeHostsFile { @@ -87,7 +96,7 @@ struct ProbeHostEntry { impl MqttPublisher { pub fn start_from_env() -> Self { let sessions = Arc::new(rocket::tokio::sync::RwLock::new(HashMap::new())); - let online_probes = Arc::new(rocket::tokio::sync::RwLock::new(HashSet::new())); + let probe_statuses = Arc::new(rocket::tokio::sync::RwLock::new(HashMap::new())); let task_timeout_ms = task_timeout_ms_from_env(); let probe_config = Arc::new(load_probe_config(task_timeout_ms).unwrap_or_else(|error| { warn!("failed to load probe config: {error}"); @@ -110,7 +119,7 @@ impl MqttPublisher { return Self { client: None, sessions, - online_probes, + probe_statuses, probe_config, task_timeout_ms: task_timeout_ms_from_env(), }; @@ -132,7 +141,7 @@ impl MqttPublisher { let (client, mut eventloop) = AsyncClient::new(options, 100); let event_sessions = sessions.clone(); - let event_online_probes = online_probes.clone(); + let event_probe_statuses = probe_statuses.clone(); let config_client = client.clone(); let event_probe_config = probe_config.clone(); rocket::tokio::spawn(async move { @@ -149,7 +158,7 @@ impl MqttPublisher { dispatch_probe_result(&event_sessions, &publish.topic, &publish.payload) .await; dispatch_probe_status( - &event_online_probes, + &event_probe_statuses, &publish.topic, &publish.payload, ) @@ -178,7 +187,7 @@ impl MqttPublisher { Self { client: Some(client), sessions, - online_probes, + probe_statuses, probe_config, task_timeout_ms, } @@ -189,7 +198,16 @@ impl MqttPublisher { } pub async fn online_probe_count(&self) -> usize { - self.online_probes.read().await.len() + self.probe_statuses + .read() + .await + .values() + .filter(|status| status.online) + .count() + } + + pub async fn probe_statuses(&self) -> HashMap { + self.probe_statuses.read().await.clone() } pub fn probe_config(&self) -> Arc { @@ -320,17 +338,13 @@ fn parse_probe_hosts(contents: &str) -> Result }) } -async fn dispatch_probe_status( - online_probes: &Arc>>, - topic: &str, - payload: &[u8], -) { +async fn dispatch_probe_status(probe_statuses: &ProbeStatuses, topic: &str, payload: &[u8]) { let Some(probe_id) = parse_probe_status_topic(topic) else { return; }; if payload.is_empty() { - online_probes.write().await.remove(probe_id); + probe_statuses.write().await.remove(probe_id); return; } @@ -342,12 +356,23 @@ async fn dispatch_probe_status( } }; - let mut online_probes = online_probes.write().await; - if status.online { - online_probes.insert(probe_id.to_string()); - } else { - online_probes.remove(probe_id); + if status.probe_id != probe_id { + warn!( + "ignoring probe status on {topic}: payload probe_id {} does not match topic", + status.probe_id + ); + return; } + + probe_statuses.write().await.insert( + probe_id.to_string(), + ProbeStatusSnapshot { + online: status.online, + version: status.version.to_string(), + dpi_hop_v4: status.dpi_hop_v4, + dpi_hop_v6: status.dpi_hop_v6, + }, + ); } async fn dispatch_probe_result( @@ -417,3 +442,61 @@ fn task_timeout_ms_from_env() -> u64 { .and_then(|timeout| timeout.parse().ok()) .unwrap_or(15_000) } + +#[cfg(test)] +mod tests { + use super::*; + + #[rocket::async_test] + async fn status_snapshot_keeps_offline_node_metadata() { + let statuses = Arc::new(rocket::tokio::sync::RwLock::new(HashMap::new())); + + dispatch_probe_status( + &statuses, + "probe/status/v1/42", + br#"{"online":false,"probe_id":"42","version":"1.2.3","dpi_hop_v4":4,"dpi_hop_v6":6}"#, + ) + .await; + + assert_eq!( + statuses.read().await.get("42"), + Some(&ProbeStatusSnapshot { + online: false, + version: "1.2.3".to_string(), + dpi_hop_v4: Some(4), + dpi_hop_v6: Some(6), + }) + ); + } + + #[rocket::async_test] + async fn empty_retained_status_removes_snapshot() { + let statuses = Arc::new(rocket::tokio::sync::RwLock::new(HashMap::from([( + "42".to_string(), + ProbeStatusSnapshot { + online: true, + version: "1.2.3".to_string(), + dpi_hop_v4: None, + dpi_hop_v6: None, + }, + )]))); + + dispatch_probe_status(&statuses, "probe/status/v1/42", b"").await; + + assert!(statuses.read().await.is_empty()); + } + + #[rocket::async_test] + async fn mismatched_payload_id_is_ignored() { + let statuses = Arc::new(rocket::tokio::sync::RwLock::new(HashMap::new())); + + dispatch_probe_status( + &statuses, + "probe/status/v1/42", + br#"{"online":true,"probe_id":"7","version":"1.2.3"}"#, + ) + .await; + + assert!(statuses.read().await.is_empty()); + } +} diff --git a/website/src/mqtt_auth.rs b/website/src/mqtt_auth.rs index 3c7cd78..7e8e975 100644 --- a/website/src/mqtt_auth.rs +++ b/website/src/mqtt_auth.rs @@ -70,15 +70,24 @@ pub async fn auth( return Json(MqttAuthResponse::deny()); } - let reporter_id = - sqlx::query_scalar::<_, i32>("SELECT id FROM reporters WHERE token = $1 LIMIT 1") - .bind(request.password) - .fetch_optional(&**pool) - .await - .ok() - .flatten(); + let Ok(reporter_id) = request.clientid.parse::() else { + return Json(MqttAuthResponse::deny()); + }; + let authenticated = sqlx::query_scalar::<_, i32>( + "UPDATE reporters + SET last_connected_at = NOW() + WHERE id = $1 AND token = $2 + RETURNING id", + ) + .bind(reporter_id) + .bind(request.password) + .fetch_optional(&**pool) + .await + .ok() + .flatten() + .is_some(); - if reporter_id.is_some_and(|id| id.to_string() == request.clientid) { + if authenticated { Json(MqttAuthResponse::allow()) } else { Json(MqttAuthResponse::deny())