From c97104929064a9cf2efeff114629a50eb659d450 Mon Sep 17 00:00:00 2001 From: imengying Date: Wed, 9 Sep 2026 00:11:03 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96=E6=95=B0=E6=8D=AE=E6=B6=88?= =?UTF-8?q?=E8=80=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .github/workflows/release.yml | 4 + .gitignore | 1 + README.md | 4 +- agent/Cargo.lock | 58 ++++ agent/Cargo.toml | 1 + agent/src/live_batch.rs | 15 +- agent/src/main.rs | 238 +++++--------- backend/Cargo.lock | 28 ++ backend/Cargo.toml | 1 + backend/src/db/queries.rs | 427 ++++++++++++++++++------- backend/src/main.rs | 51 +-- backend/src/models.rs | 87 +---- backend/src/routes/admin.rs | 8 +- backend/src/routes/public.rs | 32 +- backend/src/security.rs | 40 --- backend/src/websocket/agent.rs | 291 ++++++++--------- backend/src/websocket/dashboard.rs | 8 +- backend/src/websocket/ingest.rs | 110 +++++++ backend/src/websocket/mod.rs | 8 +- frontend/src/App.tsx | 4 +- frontend/src/AppProvider.tsx | 45 ++- frontend/src/api.ts | 5 - frontend/src/components/AdminPanel.tsx | 4 +- frontend/src/components/NodeCard.tsx | 5 +- frontend/src/hooks/useNodeLatency.ts | 25 +- frontend/src/refresh.test.ts | 65 +--- frontend/src/refresh.ts | 28 -- frontend/src/transport.test.ts | 51 ++- frontend/src/transport.ts | 55 +++- scripts/agent-runtime-test.mjs | 57 +++- scripts/smoke-test.sh | 2 +- scripts/websocket-smoke.mjs | 64 ++-- shared/Cargo.lock | 162 ++++++++++ shared/Cargo.toml | 14 + shared/telemetry.rs | 347 ++++++++++++++++++++ 35 files changed, 1491 insertions(+), 854 deletions(-) create mode 100644 backend/src/websocket/ingest.rs create mode 100644 shared/Cargo.lock create mode 100644 shared/Cargo.toml create mode 100644 shared/telemetry.rs diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 3cb9723..e547495 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -122,6 +122,10 @@ jobs: run: cargo fmt --manifest-path backend/Cargo.toml -- --check - name: Check Agent formatting run: cargo fmt --manifest-path agent/Cargo.toml -- --check + - name: Check telemetry formatting + run: cargo fmt --manifest-path shared/Cargo.toml -- --check + - name: Test telemetry codec + run: CARGO_TARGET_DIR=target/backend cargo test --locked --manifest-path shared/Cargo.toml - name: Lint backend run: CARGO_TARGET_DIR=target/backend cargo clippy --locked --manifest-path backend/Cargo.toml --all-targets -- -D warnings - name: Lint Agent diff --git a/.gitignore b/.gitignore index 767e9fd..1bec654 100644 --- a/.gitignore +++ b/.gitignore @@ -4,6 +4,7 @@ /backend/target/ /backend/build/ /agent/target/ +/shared/target/ /frontend/dist/ /frontend/admin-dist/ /release/ diff --git a/README.md b/README.md index 6215455..15eae04 100644 --- a/README.md +++ b/README.md @@ -121,7 +121,9 @@ Linux systemd 下，请通过 SSH 或本机终端重装、手动更新或卸载 远程执行是异步批量任务：只向当前连接的 Agent 下发命令，离线节点直接报告失败，重连后不会自动重发旧命令。多行脚本保留原始内容，执行结束后一次性返回输出；Agent 对单条命令设有 10 分钟执行上限。离开后台或退出登录不会停止已下发的命令。结果每 2 秒查询一次，持续 1 分钟后暂停自动刷新，可点击“刷新结果”继续查询；暂停查询不代表命令执行失败或已停止。 -实时上/下行速率由 Agent 按“收发字节增量 / 实际采样秒数”计算，前台每秒合并展示最新样本，无新样本时保持原值。节点实时采样间隔默认为 1 秒；设置更长间隔时不会凭空生成每秒数据。历史保存间隔独立，默认 60 秒。 +实时上/下行速率由 Agent 按“收发字节增量 / 实际采样秒数”计算，默认每 3 秒采集并上传一次，可设置为 3–60 秒。前台每秒合并展示已收到的最新样本，无新样本时保持原值。历史保存间隔独立，默认 60 秒。 + +Agent 和前台的监控数据使用 gzip 压缩的二进制 WebSocket 消息，机器静态信息仅在连接建立或发生变化时上传。正常连接只发送新增样本，后端暂存后按历史保存间隔批量写入数据库并确认；Agent 保留未确认样本，断线后补传。页面访问不会改变节点上报频率。 节点价格最低为 `0`，表示免费；不再接受负价格。是否隐藏节点仅由“隐藏节点”开关决定。 diff --git a/agent/Cargo.lock b/agent/Cargo.lock index 6a2f574..4f7280e 100644 --- a/agent/Cargo.lock +++ b/agent/Cargo.lock @@ -2,6 +2,12 @@ # It is not intended for manual editing. version = 4 +[[package]] +name = "adler2" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa" + [[package]] name = "anstyle" version = "1.0.14" @@ -145,6 +151,15 @@ dependencies = [ "libc", ] +[[package]] +name = "crc32fast" +version = "1.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8498c871161e1742aaa9d52551b2d6ebdd4c3d45a3be423e3728f33b955be550" +dependencies = [ + "cfg-if", +] + [[package]] name = "crypto-common" version = "0.2.2" @@ -209,6 +224,17 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d45db016d36b838f563236e9193d0ee6ce38f3f68b6c94e914b4929c96bbb890" +[[package]] +name = "flate2" +version = "1.1.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6e634e2e0ebac1ee034020da1ca582e17ffe4e0f5e985823721e168928136dcb" +dependencies = [ + "crc32fast", + "miniz_oxide", + "zlib-rs", +] + [[package]] name = "form_urlencoded" version = "1.2.2" @@ -434,12 +460,23 @@ version = "2.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" +[[package]] +name = "miniz_oxide" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b63fbc4a50860e98e7b2aa7804ded1db5cbc3aff9193adaff57a6931bf7c4b4c" +dependencies = [ + "adler2", + "simd-adler32", +] + [[package]] name = "nodeflare-agent" version = "1.0.0" dependencies = [ "clap", "httpdate", + "nodeflare-telemetry", "serde", "serde_json", "sha2", @@ -449,6 +486,15 @@ dependencies = [ "url", ] +[[package]] +name = "nodeflare-telemetry" +version = "1.0.0" +dependencies = [ + "flate2", + "serde", + "serde_json", +] + [[package]] name = "ntapi" version = "0.4.3" @@ -671,6 +717,12 @@ version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" +[[package]] +name = "simd-adler32" +version = "0.3.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3a219298ac11a56ea9a6d2120044824d6f01aeb034955e7af7bc16858527deea" + [[package]] name = "smallvec" version = "1.16.0" @@ -1218,6 +1270,12 @@ dependencies = [ "syn 3.0.4", ] +[[package]] +name = "zlib-rs" +version = "0.6.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "34b31d188d9d685a4f9c7b46d6e36631b07058d2cfe190267adce54dc230bf12" + [[package]] name = "zmij" version = "1.0.23" diff --git a/agent/Cargo.toml b/agent/Cargo.toml index 12f0f9b..455e270 100644 --- a/agent/Cargo.toml +++ b/agent/Cargo.toml @@ -10,6 +10,7 @@ repository = "https://github.com/elysia62/NodeFlare" clap = { version = "4.6", default-features = false, features = ["derive", "std", "help", "usage", "error-context", "suggestions"] } serde = { version = "1", features = ["derive"] } serde_json = "1" +nodeflare-telemetry = { path = "../shared" } sha2 = "0.11" httpdate = "1" ureq = { version = "3.4", default-features = false, features = ["json", "rustls"] } diff --git a/agent/src/live_batch.rs b/agent/src/live_batch.rs index 75aa1d2..2b5ab5b 100644 --- a/agent/src/live_batch.rs +++ b/agent/src/live_batch.rs @@ -2,7 +2,7 @@ use std::collections::VecDeque; use super::Report; -pub(crate) const MAX_LIVE_BATCH_BYTES: usize = 768 * 1024; +pub(crate) const MAX_LIVE_BATCH_BYTES: usize = nodeflare_telemetry::MAX_BATCH_BYTES; pub(crate) fn batch_len(queue: &VecDeque) -> usize { let mut encoded_bytes = 192_usize; @@ -13,7 +13,7 @@ pub(crate) fn batch_len(queue: &VecDeque) -> usize { }; let next_bytes = encoded_bytes .saturating_add(report_bytes.len()) - .saturating_add(usize::from(count > 0)); + .saturating_add(256); if next_bytes > MAX_LIVE_BATCH_BYTES { break; } @@ -43,8 +43,9 @@ mod tests { let count = batch_len(&queue); assert!((1..10).contains(&count)); let payload = crate::live_update_payload( - &queue.iter().take(count).cloned().collect::>(), + queue.iter().take(count).cloned().collect::>(), true, + &mut None, ) .unwrap(); assert!(payload.len() <= MAX_LIVE_BATCH_BYTES); @@ -75,7 +76,7 @@ mod tests { if batch.is_empty() { break; } - let message = crate::live_update_payload(&batch, true).unwrap(); + let message = crate::live_update_payload(batch.clone(), true, &mut None).unwrap(); assert!(message.len() <= MAX_LIVE_BATCH_BYTES); persisted = batch.last().unwrap().timestamp; received.extend(batch.into_iter().map(|report| report.timestamp)); @@ -95,10 +96,10 @@ mod tests { let reports = crate::live_batch_after(&queue, 100); assert_eq!(reports.first().unwrap().timestamp, 101); for persist in [true, false] { - let message = crate::live_update_payload(&reports, persist).unwrap(); - let value: serde_json::Value = serde_json::from_str(&message).unwrap(); + let message = crate::live_update_payload(reports.clone(), persist, &mut None).unwrap(); + let value: serde_json::Value = nodeflare_telemetry::decode(&message).unwrap(); assert_eq!(value["persist"], persist); - assert_eq!(value.as_object().unwrap().len(), 4); + assert_eq!(value.as_object().unwrap().len(), 2); } } } diff --git a/agent/src/main.rs b/agent/src/main.rs index 40b7f21..df4b956 100644 --- a/agent/src/main.rs +++ b/agent/src/main.rs @@ -1,5 +1,7 @@ mod live_batch; mod runtime_stats; +use nodeflare_telemetry as telemetry; +use telemetry::{DiskMetric, GpuMetric, LatencyResult, Report}; use std::collections::{HashMap, HashSet, VecDeque}; use std::env; @@ -63,7 +65,6 @@ const MAX_PENDING_SPOOL_BYTES: u64 = 64 * 1024 * 1024; const MAX_PENDING_SPOOL_LINE_BYTES: usize = 1024 * 1024; const LIVE_ACK_READ_TIMEOUT: Duration = Duration::from_secs(5); const LIVE_HINT_READ_TIMEOUT: Duration = Duration::from_millis(10); -const LIVE_REPORT_DIVISOR: u64 = 15; const BASIC_INFO_REFRESH_INTERVAL: Duration = Duration::from_secs(5 * 60); const SLOW_METRICS_REFRESH_INTERVAL: Duration = Duration::from_secs(5); const PUBLIC_IP_REFRESH_INTERVAL: Duration = Duration::from_secs(10 * 60); @@ -122,14 +123,6 @@ struct LatencyTask { interval_seconds: u64, } -#[derive(Debug, Clone, Deserialize, Serialize)] -struct LatencyResult { - task_id: String, - timestamp: i64, - latency_ms: f64, - packet_loss: f64, -} - struct LatencyExecutor { task_tx: mpsc::Sender, result_rx: mpsc::Receiver, @@ -383,10 +376,7 @@ impl LiveSender { stats: Arc, ) -> Result { let pending = Arc::new((Mutex::new(VecDeque::new()), Condvar::new())); - let send_interval = Arc::new(Mutex::new(live_batch_interval( - config.report_interval, - config.collect_interval, - ))); + let send_interval = Arc::new(Mutex::new(live_batch_interval(config.collect_interval))); let persisted_through = Arc::new(AtomicI64::new(0)); let remote_config = Arc::new(Mutex::new(None)); let endpoint = live_endpoint(&config.endpoint)?; @@ -433,9 +423,9 @@ impl LiveSender { } } - fn set_send_interval(&self, report_interval: u64, collect_interval: u64) { + fn set_send_interval(&self, collect_interval: u64) { if let Ok(mut interval) = self.send_interval.lock() { - *interval = live_batch_interval(report_interval, collect_interval); + *interval = live_batch_interval(collect_interval); } self.pending.1.notify_one(); } @@ -452,78 +442,8 @@ impl LiveSender { } } -fn live_batch_interval(report_interval: u64, collect_interval: u64) -> Duration { - let seconds = report_interval - .clamp(15, 3600) - .div_ceil(LIVE_REPORT_DIVISOR) - .clamp(1, 60) - .max(collect_interval.clamp(1, 60)); - Duration::from_secs(seconds) -} - -#[derive(Debug, Default, Clone, Deserialize, Serialize)] -struct DiskMetric { - name: String, - mount_point: String, - used: i64, - total: i64, - read_bps: f64, - write_bps: f64, - read_iops: f64, - write_iops: f64, - await_ms: f64, - utilization: f64, -} - -#[derive(Debug, Clone, Deserialize, Serialize)] -struct GpuMetric { - model: String, - usage: Option, - memory_used: i64, - memory_total: i64, -} - -#[derive(Debug, Default, Clone, Deserialize, Serialize)] -struct Report { - timestamp: i64, - cpu: f64, - load1: f64, - load5: f64, - load15: f64, - mem_used: i64, - mem_total: i64, - swap_used: i64, - swap_total: i64, - disk_used: i64, - disk_total: i64, - net_in: f64, - net_out: f64, - net_rx_total: i64, - net_tx_total: i64, - uptime: i64, - processes: i64, - tcp_connections: i64, - udp_connections: i64, - cpu_cores: i64, - cpu_model: String, - os: String, - kernel: String, - arch: String, - virtualization: String, - gpu_usage: f64, - gpu_model: String, - agent_version: String, - ip_v4: String, - ip_v6: String, - disk_read_bps: f64, - disk_write_bps: f64, - disk_read_iops: f64, - disk_write_iops: f64, - disk_await_ms: f64, - disk_utilization: f64, - disks: Vec, - gpus: Vec, - latency_results: Vec, +fn live_batch_interval(collect_interval: u64) -> Duration { + Duration::from_secs(collect_interval.clamp(telemetry::MIN_COLLECT_INTERVAL, 60)) } #[derive(Debug, Default, Clone)] @@ -1720,7 +1640,7 @@ fn runtime_config(options: &CliOptions) -> Result { token, endpoint: endpoint.trim_end_matches('/').to_string(), report_interval: interval, - collect_interval: 1, + collect_interval: telemetry::MIN_COLLECT_INTERVAL, network_interface: String::new(), agent_mirror: String::new(), auto_update: true, @@ -1737,12 +1657,6 @@ fn live_endpoint(endpoint: &str) -> Result { Ok(url.to_string()) } -fn report_batch_id(reports: &[Report]) -> String { - let encoded = serde_json::to_vec(reports).unwrap_or_default(); - let digest = Sha256::digest(encoded); - digest.iter().map(|byte| format!("{byte:02x}")).collect() -} - type LiveSocket = tungstenite::WebSocket>; fn connect_live(endpoint: &str, token: &str) -> Result<(LiveSocket, Option)> { @@ -1853,10 +1767,6 @@ struct LiveAck { persisted_through_ts: i64, #[serde(rename = "nextPersistAfterMs")] next_persist_after_ms: u64, - #[serde(rename = "nextWssReportAfterMs")] - next_wss_report_after_ms: u64, - #[serde(rename = "realtimeHint")] - realtime_hint: bool, } #[derive(Debug, Deserialize)] @@ -2463,11 +2373,6 @@ fn flush_remote_results(socket: &mut LiveSocket, executor: &mut RemoteExecutor) Ok(()) } -fn ack_wss_interval(ack: &LiveAck) -> Duration { - Duration::from_millis(ack.next_wss_report_after_ms) - .clamp(Duration::from_secs(1), Duration::from_secs(3600)) -} - fn ack_persist_interval(ack: &LiveAck) -> Duration { Duration::from_millis(ack.next_persist_after_ms) .clamp(Duration::from_secs(1), Duration::from_secs(3600)) @@ -2583,7 +2488,7 @@ fn wait_for_live_ack( } set_live_read_timeout(socket, Some(remaining))?; match read_live_ack(socket)? { - LiveRead::Ack(ack) if !ack.realtime_hint => { + LiveRead::Ack(ack) => { return Ok(LiveRead::Ack(ack)); } LiveRead::Config(config) => { @@ -2598,24 +2503,30 @@ fn wait_for_live_ack( remote_executor.acknowledge(&task_id); } LiveRead::Closed => return Ok(LiveRead::Closed), - LiveRead::Ack(_) | LiveRead::Pending => {} + LiveRead::Pending => {} } } } -fn live_update_payload(reports: &[Report], persist: bool) -> Result { - let message = serde_json::json!({ - "type": "update", - "batchId": report_batch_id(reports), - "samples": reports, - "persist": persist, - }); - Ok(serde_json::to_string(&message)?) +fn live_update_payload( + reports: Vec, + persist: bool, + info: &mut Option, +) -> Result> { + let mut next_info = info.clone(); + let payload = telemetry::encode(&telemetry::Update::from_reports( + reports, + persist, + &mut next_info, + ))?; + *info = next_info; + Ok(payload) } fn observe_persisted_through(target: &AtomicI64, ack: &LiveAck) { - let _ = ack.persisted; - target.fetch_max(ack.persisted_through_ts.max(0), Ordering::AcqRel); + if ack.persisted && !ack.persistence_error { + target.fetch_max(ack.persisted_through_ts.max(0), Ordering::AcqRel); + } } fn prune_live_queue(pending: &Arc<(Mutex>, Condvar)>, persisted_through: i64) { @@ -2647,9 +2558,11 @@ fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) { stats, } = worker; let mut socket: Option = None; - let mut wss_interval = configured_interval - .lock() - .map_or(Duration::from_secs(1), |interval| *interval); + let mut wss_interval = configured_interval.lock().map_or( + Duration::from_secs(telemetry::MIN_COLLECT_INTERVAL), + |interval| *interval, + ); + let mut info = None; let mut accepted_through = persisted_through.load(Ordering::Acquire); let mut next_send_at = Instant::now(); let mut next_probe_at = Instant::now(); @@ -2671,6 +2584,7 @@ fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) { continue; } socket = Some(connected); + info = None; remote_executor.reset_delivery(); accepted_through = persisted_through.load(Ordering::Acquire); next_send_at = Instant::now(); @@ -2700,17 +2614,16 @@ fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) { let (batch, persist, more_pending) = loop { let now = Instant::now(); let persisted = persisted_through.load(Ordering::Acquire); - if now >= next_probe_at { - let persistence_batch = live_batch_after(&queue, persisted); - if !persistence_batch.is_empty() { - let last = persistence_batch.last().unwrap().timestamp; - let more_pending = queue.iter().any(|report| report.timestamp > last); - break (persistence_batch, true, more_pending); - } - } let unsent = live_batch_after(&queue, accepted_through); - if !unsent.is_empty() && now >= next_send_at { - break (unsent, false, false); + let persist = + now >= next_probe_at && queue.iter().any(|report| report.timestamp > persisted); + if persist || (!unsent.is_empty() && now >= next_send_at) { + let last = unsent + .last() + .map_or(accepted_through, |report| report.timestamp); + let more_pending = queue.iter().any(|report| report.timestamp > last); + // A commit can be empty: samples already sent remain buffered by the server. + break (unsent, persist || more_pending, more_pending); } let wake_at = if !unsent.is_empty() { @@ -2738,10 +2651,6 @@ fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) { } else { match read_live_ack(connected) { Ok(LiveRead::Closed) => drop_socket = true, - Ok(LiveRead::Ack(ack)) if ack.realtime_hint => { - wss_interval = ack_wss_interval(&ack); - next_send_at = Instant::now() + wss_interval; - } Ok(LiveRead::Ack(ack)) => { observe_persisted_through(&persisted_through, &ack); if ack.persistence_error { @@ -2750,8 +2659,6 @@ fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) { } else { prune_live_queue(&pending, ack.persisted_through_ts); accepted_through = accepted_through.max(ack.persisted_through_ts); - wss_interval = ack_wss_interval(&ack); - next_send_at = Instant::now() + wss_interval; next_probe_at = Instant::now() + ack_persist_interval(&ack); } } @@ -2792,10 +2699,15 @@ fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) { drop(queue); if let Ok(interval) = configured_interval.lock() { - wss_interval = (*interval).clamp(Duration::from_secs(1), Duration::from_secs(60)); + wss_interval = (*interval).clamp( + Duration::from_secs(telemetry::MIN_COLLECT_INTERVAL), + Duration::from_secs(60), + ); } - let payload = match live_update_payload(&batch, persist) { + let batch_last = batch.last().map(|report| report.timestamp); + let batch_len = batch.len(); + let payload = match live_update_payload(batch, persist, &mut info) { Ok(payload) => payload, Err(error) => { eprintln!("live payload encode failed: {error}"); @@ -2805,7 +2717,7 @@ fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) { }; let sent = socket .as_mut() - .is_some_and(|socket| socket.send(Message::Text(payload.into())).is_ok()); + .is_some_and(|socket| socket.send(Message::Binary(payload.into())).is_ok()); if !sent { socket = None; accepted_through = persisted_through.load(Ordering::Acquire); @@ -2814,7 +2726,14 @@ fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) { thread::sleep(LIVE_RECONNECT_DELAY); continue; } - stats.live_batch_sent(batch.len()); + stats.live_batch_sent(batch_len); + if let Some(timestamp) = batch_last { + accepted_through = accepted_through.max(timestamp); + } + next_send_at = Instant::now() + wss_interval; + if !persist { + continue; + } let mut drop_socket = false; if let Some(socket) = socket.as_mut() { @@ -2827,18 +2746,11 @@ fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) { drop_socket = true; } else { prune_live_queue(&pending, ack.persisted_through_ts); - if let Some(timestamp) = batch.iter().map(|report| report.timestamp).max() { - accepted_through = accepted_through.max(timestamp); - } accepted_through = accepted_through.max(ack.persisted_through_ts); - wss_interval = ack_wss_interval(&ack); - next_send_at = Instant::now() + wss_interval; next_probe_at = Instant::now() + ack_persist_interval(&ack); - if persist - && more_pending - && batch - .last() - .is_some_and(|report| ack.persisted_through_ts >= report.timestamp) + if more_pending + && batch_last + .is_some_and(|timestamp| ack.persisted_through_ts >= timestamp) { next_probe_at = Instant::now(); } @@ -2874,7 +2786,7 @@ fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) { fn apply_remote(config: &mut RuntimeConfig, remote: &RemoteConfig) -> bool { if valid_sample_schedule(remote.report_interval, remote.collect_interval) { config.report_interval = remote.report_interval; - config.collect_interval = remote.collect_interval; + config.collect_interval = remote.collect_interval.max(telemetry::MIN_COLLECT_INTERVAL); } config .network_interface @@ -3380,7 +3292,7 @@ fn run(options: &CliOptions, once: bool, print_only: bool) -> Result<()> { if config.collect_interval != previous_collect_interval || config.report_interval != previous_report_interval { - live.set_send_interval(config.report_interval, config.collect_interval); + live.set_send_interval(config.collect_interval); } } } @@ -3542,7 +3454,7 @@ mod tests { PUBLIC_IP_STALE_AFTER, PublicIpValue, REMOTE_RESULT_OUTPUT_BYTES, REMOTE_STREAM_OUTPUT_BYTES, RemoteExecutor, RemoteTaskJournalEntry, RemoteTaskMessage, Report, TaskResultMessage, UPDATE_CHECK_JITTER_MAX_SECONDS, ack_persist_interval, - ack_wss_interval, advance_deadline, clock_offset_from_http_date, corrected_timestamp, + advance_deadline, clock_offset_from_http_date, corrected_timestamp, execute_remote_task_with_timeout, gpu_name_from_uevent, is_public_probe_ip, live_endpoint, live_update_payload, monotonic_report_timestamp, normalized_version, parse_lspci_gpu_names, parse_pciconf_gpu_names, parse_probe_target, parse_public_ip, @@ -3693,24 +3605,20 @@ mod tests { } #[test] - fn accepts_server_realtime_ack_interval() { + fn accepts_durable_ack_interval_and_limits_upload_frequency() { let ack: LiveAck = serde_json::from_str( - r#"{"type":"ack","ts":100,"persisted":false,"persistenceError":false,"persistedThroughTs":90,"nextPersistAfterMs":60000,"nextWssReportAfterMs":5000,"realtimeHint":false}"#, + r#"{"type":"ack","ts":100,"persisted":true,"persistenceError":false,"persistedThroughTs":100,"nextPersistAfterMs":60000}"#, ) .unwrap(); - assert_eq!(ack_wss_interval(&ack), Duration::from_secs(5)); assert_eq!(ack_persist_interval(&ack), Duration::from_secs(60)); - let slow: LiveAck = serde_json::from_str( - r#"{"type":"ack","ts":100,"persisted":true,"persistenceError":false,"persistedThroughTs":100,"nextPersistAfterMs":60000,"nextWssReportAfterMs":1,"realtimeHint":true}"#, - ) - .unwrap(); - assert_eq!(ack_wss_interval(&slow), Duration::from_secs(1)); - assert!(slow.realtime_hint); let idle: LiveAck = serde_json::from_str( - r#"{"type":"ack","ts":100,"persisted":true,"persistenceError":false,"persistedThroughTs":100,"nextPersistAfterMs":120000,"nextWssReportAfterMs":3600000,"realtimeHint":false}"#, + r#"{"type":"ack","ts":100,"persisted":true,"persistenceError":false,"persistedThroughTs":100,"nextPersistAfterMs":120000}"#, ) .unwrap(); assert_eq!(ack_persist_interval(&idle), Duration::from_secs(120)); + assert_eq!(super::live_batch_interval(1), Duration::from_secs(3)); + assert_eq!(super::live_batch_interval(3), Duration::from_secs(3)); + assert_eq!(super::live_batch_interval(15), Duration::from_secs(15)); } #[cfg(unix)] @@ -3873,7 +3781,7 @@ mod tests { #[test] fn encodes_realtime_samples_as_a_batch() { let payload = live_update_payload( - &[ + vec![ Report { timestamp: 10, cpu: 20.0, @@ -3886,13 +3794,13 @@ mod tests { }, ], false, + &mut None, ) .unwrap(); - let value: serde_json::Value = serde_json::from_str(&payload).unwrap(); - assert_eq!(value["type"], "update"); + let value: serde_json::Value = nodeflare_telemetry::decode(&payload).unwrap(); assert_eq!(value["persist"], false); assert_eq!(value["samples"].as_array().unwrap().len(), 2); - assert_eq!(value["samples"][1]["cpu"], 30.0); + assert_eq!(value["samples"][1]["metrics"]["cpu"], 30.0); } #[test] diff --git a/backend/Cargo.lock b/backend/Cargo.lock index df39c52..82fb18f 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -2,6 +2,12 @@ # It is not intended for manual editing. version = 4 +[[package]] +name = "adler2" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa" + [[package]] name = "aho-corasick" version = "1.1.5" @@ -538,6 +544,8 @@ version = "1.1.10" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6e634e2e0ebac1ee034020da1ca582e17ffe4e0f5e985823721e168928136dcb" dependencies = [ + "crc32fast", + "miniz_oxide", "zlib-rs", ] @@ -1155,6 +1163,16 @@ version = "0.3.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" +[[package]] +name = "miniz_oxide" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b63fbc4a50860e98e7b2aa7804ded1db5cbc3aff9193adaff57a6931bf7c4b4c" +dependencies = [ + "adler2", + "simd-adler32", +] + [[package]] name = "mio" version = "1.2.3" @@ -1178,6 +1196,7 @@ dependencies = [ "futures-util", "hex", "ipnet", + "nodeflare-telemetry", "pbkdf2", "rand 0.10.2", "reqwest", @@ -1200,6 +1219,15 @@ dependencies = [ "zip", ] +[[package]] +name = "nodeflare-telemetry" +version = "1.0.0" +dependencies = [ + "flate2", + "serde", + "serde_json", +] + [[package]] name = "nu-ansi-term" version = "0.50.3" diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 62b39f2..17df624 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -17,6 +17,7 @@ tokio-util = { version = "0.7", features = ["io"] } tower-http = { version = "0.7.1", features = ["trace"] } serde = { version = "1", features = ["derive"] } serde_json = "1" +nodeflare-telemetry = { path = "../shared" } sqlx = { version = "0.9", default-features = false, features = ["runtime-tokio", "tls-rustls-ring", "any", "sqlite-bundled", "postgres", "migrate", "macros"] } toml = "1.1.5" tracing = "0.1" diff --git a/backend/src/db/queries.rs b/backend/src/db/queries.rs index 8efa6d0..73d2a08 100644 --- a/backend/src/db/queries.rs +++ b/backend/src/db/queries.rs @@ -43,7 +43,6 @@ pub struct AgentIdentity { pub server_id: String, pub hidden: bool, pub report_interval: i64, - pub collect_interval: i64, pub reset_day: i64, pub rx_correction: i64, pub tx_correction: i64, @@ -58,10 +57,10 @@ pub struct PersistResult { } #[derive(Debug, Clone, Default)] -struct TrafficState { +pub(crate) struct TrafficState { cycle_key: i64, reset_day: i64, - timestamp: i64, + pub timestamp: i64, raw_rx: i64, raw_tx: i64, used_rx: i64, @@ -276,6 +275,14 @@ fn aggregate_history(reports: &[AgentReport], interval: i64) -> Vec } pub async fn list_servers(db: &Database, include_hidden: bool) -> Result> { + list_servers_with_live(db, include_hidden, &HashMap::new()).await +} + +pub async fn list_servers_with_live( + db: &Database, + include_hidden: bool, + live: &HashMap, +) -> Result> { let statement = if include_hidden { "SELECT s.id, s.name, s.region, s.group_name, s.tags, s.hidden, s.expires_at, \ s.traffic_limit, s.traffic_limit_type, s.price, s.billing_cycle, s.currency, \ @@ -296,17 +303,26 @@ pub async fn list_servers(db: &Database, include_hidden: bool) -> Result = row.try_get("latest_json")?; - let report = latest_json + let persisted_report = latest_json .as_deref() .filter(|value| !value.is_empty() && *value != "{}") .and_then(|value| serde_json::from_str::(value).ok()); - let value = |read: fn(&AgentReport) -> f64| report.as_ref().map(read); - let integer = |read: fn(&AgentReport) -> i64| report.as_ref().map(read); + let report = live + .get(&id) + .filter(|current| { + current.timestamp + >= persisted_report + .as_ref() + .map_or(0, |report| report.timestamp) + }) + .or(persisted_report.as_ref()); + let value = |read: fn(&AgentReport) -> f64| report.map(read); + let integer = |read: fn(&AgentReport) -> i64| report.map(read); let text = |read: fn(&AgentReport) -> &String| report.as_ref().map(|r| read(r).clone()); Ok(ServerView { id: id.clone(), @@ -328,7 +344,9 @@ pub async fn list_servers(db: &Database, include_hidden: bool) -> Result("collect_interval")? + .max(nodeflare_telemetry::MIN_COLLECT_INTERVAL as i64), rx_correction: row.try_get("rx_correction")?, tx_correction: row.try_get("tx_correction")?, agent_mirror: row.try_get("agent_mirror")?, @@ -544,7 +562,7 @@ pub async fn server_name(db: &Database, id: &str) -> Result> { pub async fn agent_identity(db: &Database, token: &str) -> Result> { let token_hash = auth::token_hash(token); let row = sqlx::query(db.sql( - "SELECT id, hidden, report_interval, collect_interval, reset_day, rx_correction, \ + "SELECT id, hidden, report_interval, reset_day, rx_correction, \ tx_correction FROM servers WHERE token_hash=? OR EXISTS (\ SELECT 1 FROM server_install_tokens WHERE server_id=servers.id AND token_hash=?\ )", @@ -558,7 +576,6 @@ pub async fn agent_identity(db: &Database, token: &str) -> Result("hidden")? != 0, report_interval: row.try_get("report_interval")?, - collect_interval: row.try_get("collect_interval")?, reset_day: row.try_get("reset_day")?, rx_correction: row.try_get("rx_correction")?, tx_correction: row.try_get("tx_correction")?, @@ -582,7 +599,7 @@ pub async fn agent_config(db: &Database, id: &str) -> Result("report_interval")?, - "collect_interval": row.try_get::("collect_interval")?, + "collect_interval": row.try_get::("collect_interval")?.max(nodeflare_telemetry::MIN_COLLECT_INTERVAL as i64), "network_interface": row.try_get::("network_interface")?, "agent_mirror": row.try_get::("agent_mirror")?, "auto_update": row.try_get::("auto_update")? != 0, @@ -596,7 +613,6 @@ pub async fn save_agent_batch( batch_id: &str, reports: &[AgentReport], remote_ip: &str, - persist: bool, ) -> Result { let mut reports = reports.to_vec(); reports.sort_by_key(|report| report.timestamp); @@ -615,16 +631,12 @@ pub async fn save_agent_batch( anyhow::bail!("report batch exceeds 720 samples"); } - let needs_write = persist - || reports - .iter() - .any(|report| !report.latency_results.is_empty()); - let mut transaction = if db.is_postgres() || !needs_write { + let mut transaction = if db.is_postgres() { db.pool().begin().await? } else { db.pool().begin_with("BEGIN IMMEDIATE").await? }; - if db.is_postgres() && needs_write { + if db.is_postgres() { sqlx::query("SELECT id FROM servers WHERE id=$1 FOR UPDATE") .bind(&identity.server_id) .fetch_one(&mut *transaction) @@ -666,29 +678,7 @@ pub async fn save_agent_batch( for report in &mut reports { apply_traffic(report, &mut traffic, identity); } - let latest = reports.last().context("report batch became empty")?; let report_interval = identity.report_interval.clamp(15, 3600); - let next_persist_at = persisted_before.saturating_add(report_interval); - // Only a complete persistence batch may advance the durable watermark. - if !persist { - let next_persist_after_ms = - next_persist_at.saturating_sub(latest.timestamp).max(1) as u64 * 1000; - if reports - .iter() - .any(|report| !report.latency_results.is_empty()) - { - save_latency_rows(db, &mut transaction, &identity.server_id, &reports).await?; - transaction.commit().await?; - } else { - transaction.rollback().await?; - } - return Ok(PersistResult { - reports, - persisted: false, - persisted_through: persisted_before, - next_persist_after_ms, - }); - } let history_rows = aggregate_history(&reports, report_interval); save_history_rows(db, &mut transaction, &identity.server_id, &history_rows).await?; @@ -731,7 +721,7 @@ pub async fn save_agent_batch( }) } -fn valid_agent_report(report: &AgentReport, current: i64) -> bool { +pub(crate) fn valid_agent_report(report: &AgentReport, current: i64) -> bool { fn finite_between(value: f64, minimum: f64, maximum: f64) -> bool { value.is_finite() && (minimum..=maximum).contains(&value) } @@ -849,7 +839,18 @@ async fn load_traffic_state( .map_err(Into::into) } -fn apply_traffic(report: &mut AgentReport, state: &mut TrafficState, identity: &AgentIdentity) { +pub(crate) async fn agent_traffic_state(db: &Database, server_id: &str) -> Result { + let mut transaction = db.pool().begin().await?; + let state = load_traffic_state(db, &mut transaction, server_id).await?; + transaction.rollback().await?; + Ok(state) +} + +pub(crate) fn apply_traffic( + report: &mut AgentReport, + state: &mut TrafficState, + identity: &AgentIdentity, +) { if report.timestamp <= state.timestamp { report.net_rx_total = state.used_rx.saturating_add(identity.rx_correction).max(0); report.net_tx_total = state.used_tx.saturating_add(identity.tx_correction).max(0); @@ -1171,12 +1172,13 @@ pub async fn history(db: &Database, server_id: &str, hours: i64) -> Result=? \ + AND timestamp>=? AND last_timestamp>=? \ ) samples GROUP BY bucket_timestamp ORDER BY bucket_timestamp", )) .bind(bucket) .bind(bucket) .bind(server_id) + .bind(since.saturating_sub(3600)) .bind(since) .fetch_all(db.pool()) .await?; @@ -1327,9 +1329,23 @@ fn days_in_month(year: i64, month: i64) -> i64 { } } -async fn latest_latency_map(db: &Database) -> Result>> { +async fn latest_latency_map( + db: &Database, + live: &HashMap, +) -> Result>> { + let mut live_latency = HashMap::<(&str, &str), &AgentLatencyResult>::new(); + for (server_id, report) in live { + for value in &report.latency_results { + let current = live_latency + .entry((server_id, &value.task_id)) + .or_insert(value); + if value.timestamp > current.timestamp { + *current = value; + } + } + } let rows = sqlx::query( - "SELECT a.server_id, t.id AS task_id, t.name, t.task_type, t.target, t.port, \ + "SELECT a.server_id, a.assigned_at, t.id AS task_id, t.name, t.task_type, t.target, t.port, \ lr.timestamp, lr.latency_ms, lr.packet_loss FROM latency_task_servers a \ JOIN latency_tasks t ON t.id=a.task_id LEFT JOIN latency_results lr \ ON lr.task_id=a.task_id AND lr.server_id=a.server_id AND lr.timestamp=( \ @@ -1343,24 +1359,31 @@ async fn latest_latency_map(db: &Database) -> Result>::new(); for row in rows { let server_id: String = row.try_get("server_id")?; - result - .entry(server_id.clone()) - .or_default() - .push(LatencySample { - task_id: row.try_get("task_id")?, - server_id, - name: row.try_get("name")?, - task_type: row.try_get("task_type")?, - target: row.try_get("target")?, - port: row.try_get("port")?, - timestamp: row - .try_get::, _>("timestamp")? - .unwrap_or_default(), - latency_ms: row.try_get::, _>("latency_ms")?.unwrap_or(-1.0), - packet_loss: row - .try_get::, _>("packet_loss")? - .unwrap_or(-1.0), - }); + let mut point = LatencySample { + task_id: row.try_get("task_id")?, + server_id: server_id.clone(), + name: row.try_get("name")?, + task_type: row.try_get("task_type")?, + target: row.try_get("target")?, + port: row.try_get("port")?, + timestamp: row + .try_get::, _>("timestamp")? + .unwrap_or_default(), + latency_ms: row.try_get::, _>("latency_ms")?.unwrap_or(-1.0), + packet_loss: row + .try_get::, _>("packet_loss")? + .unwrap_or(-1.0), + }; + let assigned_at = row.try_get::("assigned_at")?; + if let Some(current) = live_latency + .get(&(server_id.as_str(), point.task_id.as_str())) + .filter(|value| value.timestamp >= point.timestamp && value.timestamp >= assigned_at) + { + point.timestamp = current.timestamp; + point.latency_ms = current.latency_ms; + point.packet_loss = current.packet_loss; + } + result.entry(server_id).or_default().push(point); } Ok(result) } @@ -2672,17 +2695,17 @@ mod tests { sample(start + 1, 90.0), sample(start + 2, 20.0), ]; - let first = save_agent_batch(&db, &identity, "first", &reports[..2], "", true) + let first = save_agent_batch(&db, &identity, "first", &reports[..2], "") .await .unwrap(); assert_eq!(first.persisted_through, start + 1); // Reconnect with an overlapping batch: only the unacknowledged sample is added. - let second = save_agent_batch(&db, &identity, "overlap", &reports[1..], "", true) + let second = save_agent_batch(&db, &identity, "overlap", &reports[1..], "") .await .unwrap(); assert_eq!(second.persisted_through, start + 2); for id in ["overlap", "another-retry"] { - let retry = save_agent_batch(&db, &identity, id, &reports, "", true) + let retry = save_agent_batch(&db, &identity, id, &reports, "") .await .unwrap(); assert!(retry.reports.is_empty()); @@ -2712,20 +2735,27 @@ mod tests { let reports = (0..121) .map(|offset| sample(start + offset, 25.0)) .collect::>(); - save_agent_batch(&db, &identity, "initial", &reports[..1], "", true) + save_agent_batch(&db, &identity, "initial", &reports[..1], "") .await .unwrap(); - let live = save_agent_batch(&db, &identity, "live", &reports[120..], "", false) + let mut buffer = crate::websocket::ingest::AgentBuffer::new(&db, &identity.server_id) .await .unwrap(); - assert!(!live.persisted); - assert_eq!(live.persisted_through, start); + let live = buffer + .receive(&db, &identity, "", reports[120..].to_vec(), false) + .await + .unwrap(); + assert!(live.acknowledgement.is_none()); + assert_eq!(live.latest.unwrap().timestamp, start + 120); + assert_eq!( + list_servers(&db, true).await.unwrap()[0].timestamp, + Some(start) + ); // Even a payload-size-limited batch shorter than report_interval must commit. for (index, batch) in reports[1..].chunks(3).enumerate() { - let result = - save_agent_batch(&db, &identity, &format!("batch-{index}"), batch, "", true) - .await - .unwrap(); + let result = save_agent_batch(&db, &identity, &format!("batch-{index}"), batch, "") + .await + .unwrap(); assert!(result.persisted); assert_eq!(result.persisted_through, batch.last().unwrap().timestamp); } @@ -2738,11 +2768,182 @@ mod tests { assert_eq!(count, 121); } + #[tokio::test] + async fn latest_latency_cache_respects_probe_assignment_boundaries() { + let (db, identity) = test_server().await; + let current = now(); + let task_id = create_latency_task( + &db, + &LatencyTaskInput { + name: "Cache test".into(), + task_type: "tcp".into(), + target: "example.com".into(), + port: Some(443), + interval_seconds: 60, + default_enabled: false, + server_ids: vec![identity.server_id.clone()], + }, + ) + .await + .unwrap(); + let live = HashMap::from([( + identity.server_id.clone(), + AgentReport { + timestamp: current, + latency_results: vec![AgentLatencyResult { + task_id: task_id.clone(), + timestamp: current, + latency_ms: 28.4, + packet_loss: 0.0, + }], + ..AgentReport::default() + }, + )]); + assert_eq!( + list_servers_with_live(&db, true, &live).await.unwrap()[0].latency[0].latency_ms, + 28.4 + ); + sqlx::query("UPDATE latency_task_servers SET assigned_at=? WHERE task_id=?") + .bind(current + 1) + .bind(&task_id) + .execute(db.pool()) + .await + .unwrap(); + let latest = list_servers_with_live(&db, true, &live) + .await + .unwrap() + .remove(0); + assert_eq!(latest.latency[0].timestamp, 0); + assert_eq!(latest.latency[0].latency_ms, -1.0); + } + + #[tokio::test] + async fn live_buffer_commits_without_reupload_and_replays_only_unconfirmed_samples() { + use crate::websocket::ingest::AgentBuffer; + let (db, identity) = test_server().await; + let start = now() - 120; + let first = sample(start, 10.0); + let second = sample(start + 3, 20.0); + let third = sample(start + 6, 30.0); + let mut buffer = AgentBuffer::new(&db, &identity.server_id).await.unwrap(); + let initial = buffer + .receive(&db, &identity, "", vec![first.clone()], true) + .await + .unwrap(); + assert_eq!(initial.acknowledgement.unwrap().persisted_through, start); + + let result = buffer + .receive(&db, &identity, "", vec![second.clone()], false) + .await + .unwrap(); + assert!(result.acknowledgement.is_none()); + let latest = result.latest.unwrap(); + let snapshot = HashMap::from([(identity.server_id.clone(), latest)]); + assert_eq!( + list_servers_with_live(&db, true, &snapshot).await.unwrap()[0].cpu, + Some(20.0) + ); + assert_eq!(list_servers(&db, true).await.unwrap()[0].cpu, Some(10.0)); + let duplicate = buffer + .receive(&db, &identity, "", vec![second.clone()], false) + .await + .unwrap(); + assert!(duplicate.latest.is_none()); + let committed = buffer + .receive(&db, &identity, "", Vec::new(), true) + .await + .unwrap(); + assert_eq!( + committed.acknowledgement.unwrap().persisted_through, + start + 3 + ); + assert!(committed.latest.is_none()); + + buffer + .receive(&db, &identity, "", vec![third.clone()], false) + .await + .unwrap(); + drop(buffer); + let mut reconnected = AgentBuffer::new(&db, &identity.server_id).await.unwrap(); + let replay = reconnected + .receive(&db, &identity, "", vec![first, second, third], true) + .await + .unwrap(); + let ack = replay.acknowledgement.unwrap(); + assert_eq!(ack.persisted_through, start + 6); + assert_eq!(ack.reports.len(), 1); + assert_eq!( + history(&db, &identity.server_id, 1) + .await + .unwrap() + .iter() + .map(|point| point.sample_count) + .sum::(), + 3 + ); + assert_eq!( + ack.reports[0].net_rx_total, + replay.latest.unwrap().net_rx_total + ); + } + + #[tokio::test] + async fn live_buffer_bounds_pending_samples_and_retains_all_latency_results() { + use crate::websocket::ingest::AgentBuffer; + let (db, identity) = test_server().await; + let start = now() - 2400; + let reports: Vec<_> = (0..721) + .map(|index| AgentReport { + timestamp: start + index * 3, + latency_results: vec![AgentLatencyResult { + task_id: "test-task".into(), + timestamp: start + index * 3, + latency_ms: 20.0, + packet_loss: 0.0, + }], + ..AgentReport::default() + }) + .collect(); + let mut buffer = AgentBuffer::new(&db, &identity.server_id).await.unwrap(); + let buffered = buffer + .receive(&db, &identity, "", reports[..719].to_vec(), false) + .await + .unwrap(); + assert!(buffered.acknowledgement.is_none()); + assert_eq!(buffered.latest.unwrap().latency_results.len(), 719); + let bounded = buffer + .receive(&db, &identity, "", reports[719..].to_vec(), false) + .await + .unwrap(); + assert_eq!( + bounded.acknowledgement.unwrap().persisted_through, + reports[718].timestamp + ); + assert_eq!(bounded.latest.unwrap().latency_results.len(), 2); + let committed = buffer + .receive(&db, &identity, "", Vec::new(), true) + .await + .unwrap(); + assert_eq!( + committed.acknowledgement.unwrap().persisted_through, + reports[720].timestamp + ); + assert_eq!( + history(&db, &identity.server_id, 1) + .await + .unwrap() + .iter() + .map(|point| point.sample_count) + .sum::(), + 721 + ); + } + #[tokio::test] async fn failed_persistence_rolls_back_history_and_watermark() { let (db, identity) = test_server().await; let start = now() - 60; - save_agent_batch(&db, &identity, "initial", &[sample(start, 10.0)], "", true) + save_agent_batch(&db, &identity, "initial", &[sample(start, 10.0)], "") .await .unwrap(); sqlx::query( @@ -2753,17 +2954,14 @@ mod tests { .await .unwrap(); let next = sample(start + 1, 90.0); - assert!( - save_agent_batch( - &db, - &identity, - "next", - std::slice::from_ref(&next), - "", - true - ) + let mut buffer = crate::websocket::ingest::AgentBuffer::new(&db, &identity.server_id) .await - .is_err() + .unwrap(); + assert!( + buffer + .receive(&db, &identity, "", vec![next], true) + .await + .is_err() ); let count = sqlx::query_scalar::<_, i64>( "SELECT CAST(SUM(sample_count) AS BIGINT) FROM metric_history", @@ -2782,10 +2980,11 @@ mod tests { .execute(db.pool()) .await .unwrap(); - let result = save_agent_batch(&db, &identity, "next", &[next], "", true) + let result = buffer + .receive(&db, &identity, "", Vec::new(), true) .await .unwrap(); - assert_eq!(result.persisted_through, start + 1); + assert_eq!(result.acknowledgement.unwrap().persisted_through, start + 1); } #[tokio::test] @@ -2798,7 +2997,7 @@ mod tests { sample(start + 2, 90.0), sample(start + 60, 10.0), ]; - save_agent_batch(&db, &identity, "weighted", &reports, "", true) + save_agent_batch(&db, &identity, "weighted", &reports, "") .await .unwrap(); let points = history(&db, &identity.server_id, 24).await.unwrap(); @@ -2838,7 +3037,7 @@ mod tests { let reports = (0..=120) .map(|offset| sample(start + offset, if offset == 60 { 10.0 } else { 90.0 })) .collect::>(); - save_agent_batch(&db, &identity, "alert-data", &reports, "", true) + save_agent_batch(&db, &identity, "alert-data", &reports, "") .await .unwrap(); for metric in ["cpu", "memory", "disk", "net_in", "net_out"] { @@ -2894,15 +3093,7 @@ mod tests { let identity = identity.clone(); let reports = reports.clone(); handles.push(tokio::spawn(async move { - save_agent_batch( - &db, - &identity, - &format!("retry-{index}"), - &reports, - "", - true, - ) - .await + save_agent_batch(&db, &identity, &format!("retry-{index}"), &reports, "").await })); } for handle in handles { @@ -3031,7 +3222,7 @@ mod tests { }) .collect::>(); - let result = save_agent_batch(&db, &identity, "large-batch", &reports, "127.0.0.1", true) + let result = save_agent_batch(&db, &identity, "large-batch", &reports, "127.0.0.1") .await .unwrap(); let rows = sqlx::query_scalar::<_, i64>( @@ -3054,22 +3245,24 @@ mod tests { assert_eq!(latency_rows, 600); let persisted_at = start + 599; - let skipped = save_agent_batch( - &db, - &identity, - "realtime-only", - &[AgentReport { - timestamp: persisted_at + 1, - ..AgentReport::default() - }], - "127.0.0.1", - false, - ) - .await - .unwrap(); - assert!(!skipped.persisted); - assert_eq!(skipped.persisted_through, persisted_at); - assert_eq!(skipped.next_persist_after_ms, 59_000); + let mut buffer = crate::websocket::ingest::AgentBuffer::new(&db, &identity.server_id) + .await + .unwrap(); + let skipped = buffer + .receive( + &db, + &identity, + "127.0.0.1", + vec![AgentReport { + timestamp: persisted_at + 1, + ..AgentReport::default() + }], + false, + ) + .await + .unwrap(); + assert!(skipped.acknowledgement.is_none()); + assert_eq!(skipped.latest.unwrap().timestamp, persisted_at + 1); let due = save_agent_batch( &db, @@ -3080,7 +3273,6 @@ mod tests { ..AgentReport::default() }], "127.0.0.1", - true, ) .await .unwrap(); @@ -3131,7 +3323,6 @@ mod tests { ..AgentReport::default() }], "127.0.0.1", - true, ) .await .unwrap(); @@ -3291,7 +3482,7 @@ mod tests { async fn cleanup_is_bounded_and_keeps_current_data_and_server_state() { let (db, identity) = test_server().await; let current = now(); - save_agent_batch(&db, &identity, "latest", &[sample(current, 40.0)], "", true) + save_agent_batch(&db, &identity, "latest", &[sample(current, 40.0)], "") .await .unwrap(); let task_id = create_latency_task( diff --git a/backend/src/main.rs b/backend/src/main.rs index 5467d0b..cebccc0 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -24,12 +24,9 @@ use clap::Parser; use std::collections::HashMap; use std::net::SocketAddr; use std::sync::Arc; -use std::sync::atomic::Ordering; use tokio::sync::{Mutex, Notify, RwLock, Semaphore, broadcast, watch}; use tower_http::trace::TraceLayer; -const LIVE_REPORT_DIVISOR: i64 = 15; - pub struct AppState { pub db: db::Database, pub config: config::Config, @@ -38,6 +35,7 @@ pub struct AppState { pub http: reqwest::Client, pub agents: RwLock>, pub last_agent_reports: RwLock>, + pub live_reports: RwLock>, pub started_at: i64, pub dashboard_tx: broadcast::Sender, pub database_maintenance: Mutex<()>, @@ -47,13 +45,13 @@ pub struct AppState { pub restart_tx: watch::Sender, pub login_attempts: security::AttemptLimiter, pub sensitive_attempts: security::AttemptLimiter, - pub wake_requests: security::IntervalLimiter, pub password_verifications: Arc, } impl AppState { pub async fn disconnect_agents(&self) { let agents = std::mem::take(&mut *self.agents.write().await); + self.live_reports.write().await.clear(); for connection in agents.into_values() { let _ = connection.sender.try_send(websocket::AgentCommand::Close); } @@ -84,52 +82,12 @@ impl AppState { pub async fn disconnect_agent(&self, server_id: &str) { let connection = self.agents.write().await.remove(server_id); + self.live_reports.write().await.remove(server_id); if let Some(connection) = connection { let _ = connection.sender.try_send(websocket::AgentCommand::Close); } } - pub async fn wake_agent(&self, server_id: &str) { - let Some(connection) = self.agents.read().await.get(server_id).cloned() else { - return; - }; - connection - .live_until - .store(db::now().saturating_add(75), Ordering::Release); - let payload = serde_json::json!({ - "type": "ack", - "ts": db::now(), - "persisted": false, - "persistenceError": false, - "persistedThroughTs": 0, - "nextPersistAfterMs": connection.report_interval.clamp(15, 3600) * 1000, - "nextWssReportAfterMs": connection.collect_interval.clamp(1, 60) * 1000, - "realtimeHint": true, - }) - .to_string(); - let _ = connection - .sender - .try_send(websocket::AgentCommand::Text(payload)); - } - - pub async fn agent_wss_interval_ms(&self, server_id: &str) -> u64 { - let Some(connection) = self.agents.read().await.get(server_id).cloned() else { - return 60_000; - }; - let seconds = if connection.live_until.load(Ordering::Acquire) >= db::now() { - connection.collect_interval.clamp(1, 60) - } else { - (connection - .report_interval - .clamp(15, 3600) - .saturating_add(LIVE_REPORT_DIVISOR - 1) - / LIVE_REPORT_DIVISOR) - .clamp(1, 60) - .max(connection.collect_interval.clamp(1, 60)) - }; - seconds as u64 * 1000 - } - pub async fn send_remote_task( &self, task: &models::RemoteTaskInfo, @@ -210,6 +168,7 @@ async fn main() -> Result<()> { http, agents: RwLock::new(HashMap::new()), last_agent_reports: RwLock::new(HashMap::new()), + live_reports: RwLock::new(HashMap::new()), started_at: db::now(), dashboard_tx, database_maintenance: Mutex::new(()), @@ -227,7 +186,6 @@ async fn main() -> Result<()> { std::time::Duration::from_secs(5 * 60), std::time::Duration::from_secs(10 * 60), ), - wake_requests: security::IntervalLimiter::new(std::time::Duration::from_secs(1)), password_verifications: Arc::new(Semaphore::new(4)), }); @@ -350,7 +308,6 @@ async fn main() -> Result<()> { .route("/api/history/{id}", get(routes::public::history)) .route("/api/latency/{id}", get(routes::public::latency_history)) .route("/api/exchange-rates", get(routes::public::exchange_rates)) - .route("/api/live/wake", post(routes::public::wake_servers)) .route( "/api/turnstile/verify", post(routes::auth::verify_turnstile), diff --git a/backend/src/models.rs b/backend/src/models.rs index 6a84c18..5760fde 100644 --- a/backend/src/models.rs +++ b/backend/src/models.rs @@ -1,3 +1,7 @@ +pub use nodeflare_telemetry::{ + DiskMetric as AgentDiskMetric, GpuMetric as AgentGpuMetric, + LatencyResult as AgentLatencyResult, Report as AgentReport, +}; use serde::{Deserialize, Serialize}; use std::collections::BTreeMap; @@ -139,83 +143,6 @@ pub struct LoginSessionView { pub current: bool, } -#[derive(Debug, Clone, Default, Deserialize, Serialize)] -#[serde(default, deny_unknown_fields)] -pub struct AgentReport { - pub timestamp: i64, - pub cpu: f64, - pub load1: f64, - pub load5: f64, - pub load15: f64, - pub mem_used: i64, - pub mem_total: i64, - pub swap_used: i64, - pub swap_total: i64, - pub disk_used: i64, - pub disk_total: i64, - pub net_in: f64, - pub net_out: f64, - pub net_rx_total: i64, - pub net_tx_total: i64, - pub uptime: i64, - pub processes: i64, - pub tcp_connections: i64, - pub udp_connections: i64, - pub cpu_cores: i64, - pub cpu_model: String, - pub os: String, - pub kernel: String, - pub arch: String, - pub virtualization: String, - pub gpu_usage: f64, - pub gpu_model: String, - pub agent_version: String, - pub ip_v4: String, - pub ip_v6: String, - pub disk_read_bps: f64, - pub disk_write_bps: f64, - pub disk_read_iops: f64, - pub disk_write_iops: f64, - pub disk_await_ms: f64, - pub disk_utilization: f64, - pub disks: Vec, - pub gpus: Vec, - pub latency_results: Vec, -} - -#[derive(Debug, Clone, Default, Deserialize, Serialize)] -#[serde(default, deny_unknown_fields)] -pub struct AgentDiskMetric { - pub name: String, - pub mount_point: String, - pub used: i64, - pub total: i64, - pub read_bps: f64, - pub write_bps: f64, - pub read_iops: f64, - pub write_iops: f64, - pub await_ms: f64, - pub utilization: f64, -} - -#[derive(Debug, Clone, Default, Deserialize, Serialize)] -#[serde(default, deny_unknown_fields)] -pub struct AgentGpuMetric { - pub model: String, - pub usage: Option, - pub memory_used: i64, - pub memory_total: i64, -} - -#[derive(Debug, Clone, Default, Deserialize, Serialize)] -#[serde(default, deny_unknown_fields)] -pub struct AgentLatencyResult { - pub task_id: String, - pub timestamp: i64, - pub latency_ms: f64, - pub packet_loss: f64, -} - #[derive(Debug, Clone, Serialize)] pub struct ServerView { pub id: String, @@ -447,12 +374,6 @@ pub struct ServerBatchInput { pub ids: Vec, } -#[derive(Debug, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct WakeServersInput { - pub server_ids: Vec, -} - #[derive(Debug, Deserialize)] #[serde(deny_unknown_fields)] pub struct Enable2FaRequest { diff --git a/backend/src/routes/admin.rs b/backend/src/routes/admin.rs index 2c5e82c..f7c7cab 100644 --- a/backend/src/routes/admin.rs +++ b/backend/src/routes/admin.rs @@ -34,7 +34,8 @@ pub struct DatabaseMigrationInput { } pub async fn servers_get(State(state): State>) -> Result { - let servers = crate::db::queries::list_servers(&state.db, true) + let live = state.live_reports.read().await.clone(); + let servers = crate::db::queries::list_servers_with_live(&state.db, true, &live) .await .map_err(ApiResponse::internal)?; Ok(Json(serde_json::json!({"servers": servers})).into_response()) @@ -871,7 +872,8 @@ fn validate_server(input: &ServerInput) -> Result<(), &'static str> { } if !(1..=31).contains(&input.reset_day) || !(15..=3600).contains(&input.report_interval) - || !(1..=60).contains(&input.collect_interval) + || !(nodeflare_telemetry::MIN_COLLECT_INTERVAL as i64..=60) + .contains(&input.collect_interval) || input.collect_interval > input.report_interval || (input.report_interval + input.collect_interval - 1) / input.collect_interval > 720 { @@ -1125,7 +1127,7 @@ mod server_validation_tests { "hidden": false, "expires_at": null, "traffic_limit": 0, "traffic_limit_type": "sum", "price": 0, "billing_cycle": 30, "currency": "CNY", "auto_renewal": false, "network_interface": "", - "reset_day": 1, "report_interval": 60, "collect_interval": 1, + "reset_day": 1, "report_interval": 60, "collect_interval": 3, "rx_correction": 0, "tx_correction": 0, "agent_mirror": "", "offline_notify_disabled": false, "auto_update": true })) diff --git a/backend/src/routes/public.rs b/backend/src/routes/public.rs index 18bf132..3492f3c 100644 --- a/backend/src/routes/public.rs +++ b/backend/src/routes/public.rs @@ -1,14 +1,13 @@ -use super::{ApiResponse, bearer_or_cookie, client_ip, cookie}; +use super::{ApiResponse, bearer_or_cookie, cookie}; use crate::AppState; use crate::db::Settings; -use crate::models::{PublicConfig, WakeServersInput}; +use crate::models::PublicConfig; use axum::Json; -use axum::extract::{ConnectInfo, Path, Query, State}; -use axum::http::{HeaderMap, StatusCode}; +use axum::extract::{Path, Query, State}; +use axum::http::HeaderMap; use axum::response::{IntoResponse, Response}; use serde::Serialize; use std::collections::HashMap; -use std::net::SocketAddr; use std::sync::Arc; #[derive(Serialize)] @@ -98,7 +97,8 @@ pub async fn bootstrap( exchange_rates: None, })); } - let servers = crate::db::queries::list_servers(&state.db, false) + let live = state.live_reports.read().await.clone(); + let servers = crate::db::queries::list_servers_with_live(&state.db, false, &live) .await .map_err(ApiResponse::internal)? .into_iter() @@ -167,26 +167,6 @@ pub async fn exchange_rates(State(state): State>) -> Result>, - ConnectInfo(peer): ConnectInfo, - headers: HeaderMap, - Json(input): Json, -) -> Result { - require_dashboard(&state, &headers).await?; - if input.server_ids.len() > 500 { - return Err(ApiResponse::bad_request("服务器列表过长")); - } - let client = client_ip(&headers, peer, &state.config.trusted_proxies); - if !state.wake_requests.allow(&client) { - return Ok(StatusCode::NO_CONTENT.into_response()); - } - for server_id in input.server_ids { - state.wake_agent(&server_id).await; - } - Ok(StatusCode::NO_CONTENT.into_response()) -} - async fn require_dashboard(state: &AppState, headers: &HeaderMap) -> Result<(), ApiResponse> { let settings = crate::db::load_settings(&state.db) .await diff --git a/backend/src/security.rs b/backend/src/security.rs index 0c755ac..294db0b 100644 --- a/backend/src/security.rs +++ b/backend/src/security.rs @@ -18,38 +18,6 @@ pub struct AttemptLimiter { retention: Duration, } -pub struct IntervalLimiter { - entries: Mutex>, - minimum_interval: Duration, - retention: Duration, -} - -impl IntervalLimiter { - pub fn new(minimum_interval: Duration) -> Self { - Self { - entries: Mutex::new(HashMap::new()), - minimum_interval, - retention: minimum_interval.saturating_mul(3_600), - } - } - - pub fn allow(&self, key: &str) -> bool { - let now = Instant::now(); - let Ok(mut entries) = self.entries.lock() else { - return false; - }; - entries.retain(|_, seen| now.duration_since(*seen) <= self.retention); - if entries - .get(key) - .is_some_and(|seen| now.duration_since(*seen) < self.minimum_interval) - { - return false; - } - entries.insert(key.to_string(), now); - true - } -} - impl AttemptLimiter { pub fn new(maximum_failures: u32, window: Duration, block_for: Duration) -> Self { Self { @@ -123,12 +91,4 @@ mod tests { limiter.clear("client"); assert_eq!(limiter.retry_after("client"), None); } - - #[test] - fn interval_limiter_coalesces_repeated_requests() { - let limiter = IntervalLimiter::new(Duration::from_secs(60)); - assert!(limiter.allow("client")); - assert!(!limiter.allow("client")); - assert!(limiter.allow("other")); - } } diff --git a/backend/src/websocket/agent.rs b/backend/src/websocket/agent.rs index 0805c37..7cf5deb 100644 --- a/backend/src/websocket/agent.rs +++ b/backend/src/websocket/agent.rs @@ -8,26 +8,15 @@ use axum::extract::{ConnectInfo, State}; use axum::http::HeaderMap; use axum::response::{IntoResponse, Response}; use futures_util::{SinkExt, StreamExt}; +use nodeflare_telemetry as telemetry; use serde::Deserialize; use std::net::SocketAddr; use std::sync::Arc; -use std::sync::atomic::AtomicI64; use tokio::sync::mpsc; const MAX_AGENT_MESSAGE_BYTES: usize = 2 * 1024 * 1024; const MAX_TASK_RESULT_BYTES: usize = 1024 * 1024; -#[derive(Deserialize)] -#[serde(deny_unknown_fields)] -struct UpdateMessage { - #[serde(rename = "type")] - _message_type: String, - #[serde(rename = "batchId")] - batch_id: String, - samples: Vec, - persist: bool, -} - #[derive(Deserialize)] #[serde(deny_unknown_fields)] struct TaskResultMessage { @@ -92,14 +81,12 @@ async fn run( return; }; let (outbound_tx, mut outbound_rx) = mpsc::channel::(256); + let mut info = None; let previous = state.agents.write().await.insert( identity.server_id.clone(), AgentConnection { connection_id: connection_id.clone(), sender: outbound_tx.clone(), - report_interval: identity.report_interval, - collect_interval: identity.collect_interval, - live_until: Arc::new(AtomicI64::new(0)), }, ); if let Some(previous) = previous { @@ -107,11 +94,12 @@ async fn run( } // Recheck the token after registration in case the node changed meanwhile. + let initialized = super::ingest::AgentBuffer::new(&state.db, &identity.server_id).await; let still_authorized = matches!( crate::db::queries::agent_identity(&state.db, &token).await, Ok(Some(current)) if current.server_id == identity.server_id ); - if !still_authorized { + if !still_authorized || initialized.is_err() { let mut agents = state.agents.write().await; if agents .get(&identity.server_id) @@ -124,6 +112,7 @@ async fn run( tracing::warn!(server_id = %identity.server_id, "rejected stale Agent connection"); return; } + let mut buffer = initialized.expect("Agent buffer was initialized"); tracing::info!(server_id = %identity.server_id, remote_ip, "Agent connected"); if let Some(config) = crate::db::queries::agent_config(&state.db, &identity.server_id) @@ -166,9 +155,80 @@ async fn run( }); while let Some(message) = websocket_rx.next().await { + // Registration waits for in-flight ingestion before loading the next connection's watermark. + let connections = state.agents.read().await; + if !connections + .get(&identity.server_id) + .is_some_and(|connection| connection.connection_id == connection_id) + { + break; + } match message { + Ok(Message::Binary(bytes)) => { + let result = async { + let _activity = state + .database_activity + .write() + .map_err(|_| anyhow::anyhow!("database is under maintenance"))?; + let update: telemetry::Update = telemetry::decode(&bytes)?; + let persist = update.persist; + let reports = update.into_reports(&mut info)?; + let result = buffer + .receive(&state.db, &identity, &remote_ip, reports, persist) + .await?; + state + .last_agent_reports + .write() + .await + .insert(identity.server_id.clone(), crate::db::now()); + if let Some(acknowledgement) = result.acknowledgement { + send_ack(&outbound_tx, &acknowledgement); + if !acknowledgement.reports.is_empty() { + evaluate_alerts(&state, &identity.server_id); + } + } + if let Some(report) = result.latest { + let mut live = state.live_reports.write().await; + if live + .get(&identity.server_id) + .is_none_or(|current| current.timestamp < report.timestamp) + { + broadcast_report(&state, &identity, &report); + let mut cached = report; + if let Some(previous) = live.get(&identity.server_id) { + cached + .latency_results + .extend(previous.latency_results.iter().cloned()); + } + cached.latency_results.sort_by(|left, right| { + left.task_id + .cmp(&right.task_id) + .then(right.timestamp.cmp(&left.timestamp)) + }); + cached + .latency_results + .dedup_by(|left, right| left.task_id == right.task_id); + cached + .latency_results + .retain(|value| value.timestamp >= crate::db::now() - 7200); + cached + .latency_results + .sort_by_key(|value| std::cmp::Reverse(value.timestamp)); + cached.latency_results.truncate(4096); + live.insert(identity.server_id.clone(), cached); + } + } + Ok::<_, anyhow::Error>(()) + } + .await; + if let Err(error) = result { + tracing::warn!(%error, server_id = %identity.server_id, "Agent telemetry failed"); + send_persistence_error(&outbound_tx, 0); + break; + } + } Ok(Message::Text(text)) if text.len() <= MAX_AGENT_MESSAGE_BYTES => { - handle_text(&state, &identity, &remote_ip, &outbound_tx, &text).await; + handle_text(&state, &identity, &outbound_tx, &text).await; } Ok(Message::Ping(payload)) => { let _ = outbound_tx.try_send(AgentCommand::Pong(payload.to_vec())); @@ -193,7 +253,6 @@ async fn run( async fn handle_text( state: &Arc, identity: &AgentIdentity, - remote_ip: &str, outbound: &mpsc::Sender, text: &str, ) { @@ -205,80 +264,6 @@ async fn handle_text( return; }; match value.get("type").and_then(serde_json::Value::as_str) { - Some("update") => { - let Ok(update) = serde_json::from_value::(value) else { - send_persistence_error(outbound, 0); - return; - }; - if !valid_batch_id(&update.batch_id) { - send_persistence_error(outbound, 0); - return; - } - match crate::db::queries::save_agent_batch( - &state.db, - identity, - &update.batch_id, - &update.samples, - remote_ip, - update.persist, - ) - .await - { - Ok(result) => { - state - .last_agent_reports - .write() - .await - .insert(identity.server_id.clone(), crate::db::now()); - let next_wss_report_after_ms = - state.agent_wss_interval_ms(&identity.server_id).await; - send_ack(outbound, &result, next_wss_report_after_ms); - if !result.reports.is_empty() { - broadcast_reports(state, identity, &result.reports); - } - if result.persisted && !result.reports.is_empty() { - let state = Arc::clone(state); - let server_id = identity.server_id.clone(); - tokio::spawn(async move { - let Ok(_activity) = state.database_activity.write() else { - return; - }; - let settings = match crate::db::load_settings(&state.db).await { - Ok(settings) => settings, - Err(error) => { - tracing::error!(%error, "failed to load alert settings"); - return; - } - }; - if !settings.notification_enabled { - return; - } - let server_name = - crate::db::queries::server_name(&state.db, &server_id) - .await - .ok() - .flatten() - .unwrap_or_else(|| server_id.clone()); - if let Err(error) = crate::notify::evaluate_resource_alerts( - &state.db, - &settings, - &server_id, - &server_name, - ) - .await - { - tracing::error!(%error, server_id, "resource alert evaluation failed"); - } - state.notification_wake.notify_one(); - }); - } - } - Err(error) => { - tracing::error!(%error, server_id = %identity.server_id, "Agent batch persistence failed"); - send_persistence_error(outbound, 0); - } - } - } Some("task_result") => { let Ok(result) = serde_json::from_value::(value) else { return; @@ -335,11 +320,39 @@ async fn handle_text( } } -fn send_ack( - outbound: &mpsc::Sender, - result: &PersistResult, - next_wss_report_after_ms: u64, -) { +fn evaluate_alerts(state: &Arc, server_id: &str) { + let state = Arc::clone(state); + let server_id = server_id.to_string(); + tokio::spawn(async move { + let Ok(_activity) = state.database_activity.write() else { + return; + }; + let settings = match crate::db::load_settings(&state.db).await { + Ok(settings) => settings, + Err(error) => { + tracing::error!(%error, "failed to load alert settings"); + return; + } + }; + if !settings.notification_enabled { + return; + } + let server_name = crate::db::queries::server_name(&state.db, &server_id) + .await + .ok() + .flatten() + .unwrap_or_else(|| server_id.clone()); + if let Err(error) = + crate::notify::evaluate_resource_alerts(&state.db, &settings, &server_id, &server_name) + .await + { + tracing::error!(%error, server_id, "resource alert evaluation failed"); + } + state.notification_wake.notify_one(); + }); +} + +fn send_ack(outbound: &mpsc::Sender, result: &PersistResult) { let _ = outbound.try_send(AgentCommand::Text( serde_json::json!({ "type": "ack", @@ -348,8 +361,6 @@ fn send_ack( "persistenceError": false, "persistedThroughTs": result.persisted_through, "nextPersistAfterMs": result.next_persist_after_ms, - "nextWssReportAfterMs": next_wss_report_after_ms, - "realtimeHint": false, }) .to_string(), )); @@ -364,58 +375,43 @@ fn send_persistence_error(outbound: &mpsc::Sender, persisted: i64) "persistenceError": true, "persistedThroughTs": persisted, "nextPersistAfterMs": 5000, - "nextWssReportAfterMs": 5000, - "realtimeHint": false, }) .to_string(), )); } -fn broadcast_reports(state: &AppState, identity: &AgentIdentity, reports: &[AgentReport]) { - if identity.hidden { +fn broadcast_report(state: &AppState, identity: &AgentIdentity, report: &AgentReport) { + if identity.hidden || state.dashboard_tx.receiver_count() == 0 { return; } - let samples = reports - .iter() - .map(|report| { - let mut data = serde_json::to_value(report).unwrap_or_else(|_| serde_json::json!({})); - if let Some(object) = data.as_object_mut() { - for key in [ - "timestamp", - "cpu_model", - "os", - "kernel", - "arch", - "virtualization", - "gpu_model", - "agent_version", - "ip_v4", - "ip_v6", - ] { - object.remove(key); - } - } - serde_json::json!({"ts": report.timestamp, "data": data}) - }) - .collect::>(); - let payload = serde_json::json!({ + let mut data = serde_json::to_value(report).unwrap_or_else(|_| serde_json::json!({})); + if let Some(object) = data.as_object_mut() { + for key in [ + "timestamp", + "cpu_model", + "os", + "kernel", + "arch", + "virtualization", + "gpu_model", + "agent_version", + "ip_v4", + "ip_v6", + ] { + object.remove(key); + } + } + let payload = telemetry::encode(&serde_json::json!({ "type": "batchUpdate", "ts": crate::db::now(), - "updates": [{"serverId": identity.server_id, "samples": samples}], - }) - .to_string(); - let _ = state.dashboard_tx.send(DashboardEvent { - server_id: identity.server_id.clone(), - payload, - }); -} - -fn valid_batch_id(value: &str) -> bool { - !value.is_empty() - && value.len() <= 128 - && value - .bytes() - .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.' | b':')) + "updates": [{"serverId": identity.server_id, "samples": [{"ts": report.timestamp, "data": data}]}], + })); + if let Ok(payload) = payload { + let _ = state.dashboard_tx.send(DashboardEvent { + server_id: identity.server_id.clone(), + payload: payload.into(), + }); + } } fn valid_task_id(value: &str) -> bool { @@ -452,9 +448,6 @@ mod tests { let connection = AgentConnection { connection_id: "test".to_string(), sender, - report_interval: 60, - collect_interval: 1, - live_until: Arc::new(AtomicI64::new(0)), }; let task = RemoteTaskInfo { id: uuid::Uuid::new_v4().to_string(), @@ -493,14 +486,14 @@ mod tests { #[test] fn updates_require_an_explicit_persistence_flag() { - let mut message = serde_json::json!({"type": "update", "batchId": "batch", "samples": []}); - assert!(serde_json::from_value::(message.clone()).is_err()); + let mut message = serde_json::json!({"samples": []}); + assert!(serde_json::from_value::(message.clone()).is_err()); message["persist"] = serde_json::Value::Null; - assert!(serde_json::from_value::(message.clone()).is_err()); + assert!(serde_json::from_value::(message.clone()).is_err()); for persist in [true, false] { message["persist"] = persist.into(); assert_eq!( - serde_json::from_value::(message.clone()) + serde_json::from_value::(message.clone()) .unwrap() .persist, persist diff --git a/backend/src/websocket/dashboard.rs b/backend/src/websocket/dashboard.rs index 462ca95..609f723 100644 --- a/backend/src/websocket/dashboard.rs +++ b/backend/src/websocket/dashboard.rs @@ -86,11 +86,15 @@ async fn run( tokio::select! { update = updates.recv() => match update { Ok(update) if server_id.as_ref().is_none_or(|id| id == &update.server_id) => { - if sender.send(Message::Text(update.payload.into())).await.is_err() { + if sender.send(Message::Binary(update.payload)).await.is_err() { break; } } - Ok(_) | Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {} + Ok(_) => {} + Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => { + let _ = sender.send(Message::Close(None)).await; + break; + } Err(tokio::sync::broadcast::error::RecvError::Closed) => break, }, incoming = receiver.next() => match incoming { diff --git a/backend/src/websocket/ingest.rs b/backend/src/websocket/ingest.rs new file mode 100644 index 0000000..12e1b5d --- /dev/null +++ b/backend/src/websocket/ingest.rs @@ -0,0 +1,110 @@ +use crate::db::{ + Database, now, + queries::{self, AgentIdentity, PersistResult, TrafficState}, +}; +use crate::models::AgentReport; +use anyhow::{Result, ensure}; +use nodeflare_telemetry::{MAX_BATCH_BYTES, MAX_SAMPLES}; + +pub struct AgentBuffer { + pending: Vec, + pending_bytes: usize, + received_through: i64, + persisted_through: i64, + traffic: TrafficState, +} + +pub struct IngestResult { + pub latest: Option, + pub acknowledgement: Option, +} + +impl AgentBuffer { + pub async fn new(db: &Database, server_id: &str) -> Result { + let traffic = queries::agent_traffic_state(db, server_id).await?; + Ok(Self { + pending: Vec::new(), + pending_bytes: 0, + received_through: traffic.timestamp, + persisted_through: traffic.timestamp, + traffic, + }) + } + + pub async fn receive( + &mut self, + db: &Database, + identity: &AgentIdentity, + remote_ip: &str, + mut reports: Vec, + persist: bool, + ) -> Result { + let current = now(); + ensure!(reports.len() <= MAX_SAMPLES, "too many telemetry samples"); + ensure!( + reports.iter().all(|report| report.timestamp > 0 + && report.timestamp >= current - 7200 + && report.timestamp <= current + 300 + && queries::valid_agent_report(report, current)), + "invalid telemetry sample" + ); + reports.sort_by_key(|report| report.timestamp); + reports.dedup_by_key(|report| report.timestamp); + reports.retain(|report| report.timestamp > self.received_through); + let bytes = reports.iter().try_fold(0_usize, |size, report| { + serde_json::to_vec(report).map(|encoded| size.saturating_add(encoded.len())) + })?; + ensure!(bytes <= MAX_BATCH_BYTES, "telemetry batch is too large"); + let mut acknowledgement = None; + if !self.pending.is_empty() + && (self.pending.len() + reports.len() > MAX_SAMPLES + || self.pending_bytes.saturating_add(bytes) > MAX_BATCH_BYTES) + { + acknowledgement = Some(self.flush(db, identity, remote_ip).await?); + } + let mut latest = None; + let mut latency = Vec::new(); + for report in &reports { + let mut live = report.clone(); + queries::apply_traffic(&mut live, &mut self.traffic, identity); + self.received_through = live.timestamp; + latency.append(&mut live.latency_results); + latest = Some(live); + } + if let Some(latest) = &mut latest { + latest.latency_results = latency; + } + self.pending.extend(reports); + self.pending_bytes = self.pending_bytes.saturating_add(bytes); + if persist || self.pending_bytes >= MAX_BATCH_BYTES || self.pending.len() >= MAX_SAMPLES { + acknowledgement = Some(self.flush(db, identity, remote_ip).await?); + } + Ok(IngestResult { + latest, + acknowledgement, + }) + } + + async fn flush( + &mut self, + db: &Database, + identity: &AgentIdentity, + remote_ip: &str, + ) -> Result { + let Some(first) = self.pending.first() else { + return Ok(PersistResult { + reports: Vec::new(), + persisted: true, + persisted_through: self.persisted_through, + next_persist_after_ms: identity.report_interval.clamp(15, 3600) as u64 * 1000, + }); + }; + let batch_id = format!("{}:{}", first.timestamp, self.received_through); + let result = + queries::save_agent_batch(db, identity, &batch_id, &self.pending, remote_ip).await?; + self.persisted_through = result.persisted_through; + self.pending.clear(); + self.pending_bytes = 0; + Ok(result) + } +} diff --git a/backend/src/websocket/mod.rs b/backend/src/websocket/mod.rs index 1f9f9c0..0ca1215 100644 --- a/backend/src/websocket/mod.rs +++ b/backend/src/websocket/mod.rs @@ -1,17 +1,13 @@ pub mod agent; pub mod dashboard; +pub(crate) mod ingest; -use std::sync::Arc; -use std::sync::atomic::AtomicI64; use tokio::sync::mpsc; #[derive(Clone)] pub struct AgentConnection { pub connection_id: String, pub sender: mpsc::Sender, - pub report_interval: i64, - pub collect_interval: i64, - pub live_until: Arc, } #[derive(Debug)] @@ -24,5 +20,5 @@ pub enum AgentCommand { #[derive(Debug, Clone)] pub struct DashboardEvent { pub server_id: String, - pub payload: String, + pub payload: axum::body::Bytes, } diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index ed5f51c..8670667 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -96,7 +96,7 @@ function VerificationGate() { } function HomeView() { - const { carrierLatency, config, error, exchangeRates, liveMetrics, loading, openServer, reload, servers } = useApp(); + const { carrierLatency, config, error, exchangeRates, liveConnected, liveMetrics, loading, openServer, reload, servers } = useApp(); const [query, setQuery] = useState(""); const [group, setGroup] = useState("__all__"); const locale = config.locale; @@ -125,7 +125,7 @@ function HomeView() {
) : visible.length ? (
- {visible.map((server) => openServer(server)} />)} + {visible.map((server) => openServer(server)} />)}
) : !error ? (
{servers.length ? ui(locale, "没有匹配的节点", "No matching servers") : ui(locale, "尚未添加节点", "No servers added")}
diff --git a/frontend/src/AppProvider.tsx b/frontend/src/AppProvider.tsx index 2d1f28f..99a997d 100644 --- a/frontend/src/AppProvider.tsx +++ b/frontend/src/AppProvider.tsx @@ -13,7 +13,6 @@ import { useFavicon, useStoredAppearance, useSystemDark } from "./hooks/useBrows import { resolveBackground, themeToggle } from "./theme"; import { BOOTSTRAP_POLL_INTERVAL_MS, - createLiveFlushScheduler, createRefreshQueue, shouldSyncBootstrap, } from "./refresh"; @@ -35,6 +34,7 @@ export interface AppState { configReady: boolean; servers: Server[]; liveMetrics: LiveMetricsMap; + liveConnected: boolean; exchangeRates: ExchangeRates | null; loading: boolean; error: string; @@ -72,6 +72,8 @@ export function AppProvider({ children }: { children: ReactNode }) { const [configReady, setConfigReady] = useState(demoMode); const [servers, setServers] = useState([]); const [liveMetrics, setLiveMetrics] = useState({}); + const [liveConnected, setLiveConnected] = useState(false); + const pendingLive = useRef([]); const [clockNow, setClockNow] = useState(() => Date.now()); const [exchangeRates, setExchangeRates] = useState(null); const [loading, setLoading] = useState(true); @@ -217,12 +219,19 @@ export function AppProvider({ children }: { children: ReactNode }) { useEffect(() => { let timer: number | undefined; + const tick = () => { + setClockNow(Date.now()); + if (!pendingLive.current.length) return; + const batch = pendingLive.current; + pendingLive.current = []; + setLiveMetrics((current) => applyBatch(current, batch, serversRef.current)); + }; const sync = () => { window.clearInterval(timer); timer = undefined; if (document.hidden || navigator.onLine === false) return; - setClockNow(Date.now()); - timer = window.setInterval(() => setClockNow(Date.now()), 1_000); + tick(); + timer = window.setInterval(tick, 1_000); }; sync(); document.addEventListener("visibilitychange", sync); @@ -239,43 +248,24 @@ export function AppProvider({ children }: { children: ReactNode }) { useEffect(() => { if (!configReady || access !== "ok" || demoMode) return; let active = true; - let pending: BatchUpdate[] = []; - // Coalesce independently timed reports onto a shared one-second UI tick. - const flush = () => { - if (!pending.length) return; - const batch = pending; - pending = []; - setLiveMetrics((current) => applyBatch(current, batch, serversRef.current)); - }; - const scheduler = createLiveFlushScheduler(flush); const disconnect = connectLive({ serverId: selectedId }, { onBatch: (updates) => { - pending.push(...updates); - scheduler.schedule(); + pendingLive.current.push(...updates); }, onConnectedChange: (connected) => { liveConnectedRef.current = connected; + setLiveConnected(connected); if (!connected) { - scheduler.cancel(); - pending = []; + pendingLive.current = []; } if (active && !connected && !document.hidden && navigator.onLine !== false) { void reload(true); } }, - onWakeRequested: async () => { - const serverIds = selectedId ? [selectedId] : serversRef.current.map((server) => server.id); - for (let offset = 0; offset < serverIds.length; offset += 500) { - // The wake endpoint accepts one request per second and at most 500 nodes. - if (offset > 0) await new Promise((resolve) => window.setTimeout(resolve, 1_100)); - if (!active || document.hidden || navigator.onLine === false) return; - await api.wakeServers(serverIds.slice(offset, offset + 500)); - } - }, }); return () => { active = false; - scheduler.cancel(); + pendingLive.current = []; disconnect(); }; }, [access, configReady, reload, selectedId]); @@ -333,6 +323,7 @@ export function AppProvider({ children }: { children: ReactNode }) { configReady, servers: liveServers, liveMetrics, + liveConnected, exchangeRates, loading, error, @@ -349,7 +340,7 @@ export function AppProvider({ children }: { children: ReactNode }) { reload: () => reload(), login, verify, - }), [access, background, blur, carrierLatency, config, configReady, dark, error, exchangeRates, goHome, liveMetrics, liveServers, loading, login, openServer, reload, selectedId, toggleTheme, verify]); + }), [access, background, blur, carrierLatency, config, configReady, dark, error, exchangeRates, goHome, liveConnected, liveMetrics, liveServers, loading, login, openServer, reload, selectedId, toggleTheme, verify]); return {children}; } diff --git a/frontend/src/api.ts b/frontend/src/api.ts index 57b6d6e..0fc3477 100644 --- a/frontend/src/api.ts +++ b/frontend/src/api.ts @@ -45,11 +45,6 @@ export const api = { bootstrap: () => request("/api/bootstrap", { cache: "no-store", signal: AbortSignal.timeout(15_000) }), config: () => request("/api/config"), exchangeRates: () => request("/api/exchange-rates"), - wakeServers: (serverIds: string[]) => request("/api/live/wake", { - method: "POST", - body: JSON.stringify({ server_ids: serverIds }), - signal: AbortSignal.timeout(15_000), - }), refreshExchangeRates: () => request("/api/admin/exchange-rates/refresh", { method: "POST" }, true), adminServers: () => request<{ servers: AdminServer[] }>("/api/admin/servers", {}, true), history: (id: string, hours: number) => diff --git a/frontend/src/components/AdminPanel.tsx b/frontend/src/components/AdminPanel.tsx index 6269dcc..2bbaa86 100644 --- a/frontend/src/components/AdminPanel.tsx +++ b/frontend/src/components/AdminPanel.tsx @@ -115,7 +115,7 @@ const emptyServer: ServerInput = { network_interface: "", reset_day: 1, report_interval: 60, - collect_interval: 1, + collect_interval: 3, rx_correction: 0, tx_correction: 0, agent_mirror: "", @@ -1309,7 +1309,7 @@ export function AdminPanel({
-
+
updateForm("auto_renewal", value)} /> updateForm("auto_update", value)} /> updateForm("hidden", value)} /> updateForm("offline_notify_disabled", value)} />
diff --git a/frontend/src/components/NodeCard.tsx b/frontend/src/components/NodeCard.tsx index 138932b..3d267d9 100644 --- a/frontend/src/components/NodeCard.tsx +++ b/frontend/src/components/NodeCard.tsx @@ -94,10 +94,11 @@ function CarrierPanel({ label, rows, kind, empty }: { ); } -export function NodeCard({ server, config, liveLatencyResults, onOpen }: { +export function NodeCard({ server, config, liveLatencyResults, liveConnected, onOpen }: { server: Server; config: Config; liveLatencyResults?: LiveLatencyResult[]; + liveConnected: boolean; onOpen: () => void; }) { const threshold = config.offline_threshold_seconds; @@ -108,7 +109,7 @@ export function NodeCard({ server, config, liveLatencyResults, onOpen }: { const traffic = server.traffic_limit > 0 ? Math.min(100, (usedTraffic / server.traffic_limit) * 100) : 0; const locale = config.locale; const showCarriers = themeToggle(config, "showCarrierLatency", false); - const quality = useNodeLatency(server, config.show_latency, locale, showCarriers ? carrierSelection(config) : null, liveLatencyResults); + const quality = useNodeLatency(server, config.show_latency, locale, showCarriers ? carrierSelection(config) : null, liveLatencyResults, liveConnected); const price = formatPrice(server, locale); const remainingValue = remainingAssetValue(server.price, server.billing_cycle, server.expires_at); const showExpiryPanel = config.show_expiry || config.show_price; diff --git a/frontend/src/hooks/useNodeLatency.ts b/frontend/src/hooks/useNodeLatency.ts index 883aedd..b9427dc 100644 --- a/frontend/src/hooks/useNodeLatency.ts +++ b/frontend/src/hooks/useNodeLatency.ts @@ -39,10 +39,21 @@ export interface NodeLatencyStats { } const cache = new Map(); +const requests = new Map>(); const CACHE_TTL = 120_000; const REFRESH_INTERVAL = 120_000; const WINDOW_HOURS = 1; +function fetchHistory(id: string, signature: string) { + const key = `${id}:${signature}`; + let request = requests.get(key); + if (!request) { + request = api.latencyHistory(id, WINDOW_HOURS).finally(() => requests.delete(key)); + requests.set(key, request); + } + return request; +} + function mergePoints(...sources: LatencySample[][]): LatencySample[] { const points = new Map(); for (const source of sources) { @@ -69,11 +80,12 @@ export function useNodeLatency( locale: UiLocale, carrierSelection: CarrierSlotKey | null = null, liveResults?: LiveLatencyResult[], + liveConnected = false, ): NodeLatencyStats { const [fetched, setFetched] = useState(server.latency); const [loading, setLoading] = useState(enabled); - const taskSignature = [...new Set(server.latency.map((point) => point.task_id))].sort().join(","); + const taskSignature = JSON.stringify(server.latency.map((point) => [point.task_id, point.name, point.task_type, point.target, point.port])); useEffect(() => { setFetched(server.latency); @@ -83,6 +95,7 @@ export function useNodeLatency( if (!enabled) { setLoading(false); return; } let stopped = false; let running = false; + let loaded = false; let timer: number | undefined; const canRefresh = () => !document.hidden && navigator.onLine !== false; @@ -94,7 +107,7 @@ export function useNodeLatency( }; const schedule = () => { clearTimer(); - if (!stopped && canRefresh()) { + if (!stopped && canRefresh() && (!liveConnected || !loaded)) { timer = window.setTimeout(() => void load(true), REFRESH_INTERVAL); } }; @@ -107,10 +120,12 @@ export function useNodeLatency( if (!force && hit?.signature === taskSignature && Date.now() - hit.at < CACHE_TTL) { setFetched(hit.points); setLoading(false); + loaded = true; return; } if (!hit) setLoading(true); - const result = await api.latencyHistory(server.id, WINDOW_HOURS); + const result = await fetchHistory(server.id, taskSignature); + loaded = true; if (!stopped) { cache.set(server.id, { at: Date.now(), signature: taskSignature, points: result.points }); setFetched(result.points); @@ -134,7 +149,7 @@ export function useNodeLatency( document.addEventListener("visibilitychange", handleVisibility); window.addEventListener("online", resume); window.addEventListener("offline", pause); - if (canRefresh()) void load(false); + if (canRefresh()) void load(liveConnected); return () => { stopped = true; clearTimer(); @@ -142,7 +157,7 @@ export function useNodeLatency( window.removeEventListener("online", resume); window.removeEventListener("offline", pause); }; - }, [enabled, server.id, taskSignature]); + }, [enabled, liveConnected, server.id, taskSignature]); const points = useMemo(() => { const cutoff = Date.now() / 1000 - WINDOW_HOURS * 3600; diff --git a/frontend/src/refresh.test.ts b/frontend/src/refresh.test.ts index e49f3df..28e41e0 100644 --- a/frontend/src/refresh.test.ts +++ b/frontend/src/refresh.test.ts @@ -1,7 +1,6 @@ -import { describe, expect, spyOn, test } from "bun:test"; +import { describe, expect, test } from "bun:test"; import { BOOTSTRAP_LIVE_SYNC_INTERVAL_MS, - createLiveFlushScheduler, createRefreshQueue, hasActiveRemoteTasks, isRemoteTaskActive, @@ -39,68 +38,6 @@ describe("createRefreshQueue", () => { }); }); -describe("createLiveFlushScheduler", () => { - test("uses a fixed one-second clock, coalesces bursts and skips idle ticks", () => { - let flushes = 0; - let tick = () => {}; - const timer = 123 as unknown as ReturnType; - const start = spyOn(globalThis, "setInterval").mockImplementation(((callback: () => void, delay?: number) => { - expect(delay).toBe(1_000); - tick = callback as () => void; - return timer; - }) as typeof setInterval); - const stop = spyOn(globalThis, "clearInterval").mockImplementation(() => {}); - try { - const scheduler = createLiveFlushScheduler(() => { flushes += 1; }); - scheduler.schedule(); - scheduler.schedule(); - scheduler.schedule(); - expect(flushes).toBe(0); - expect(start).toHaveBeenCalledTimes(1); - tick(); - expect(flushes).toBe(1); - tick(); - expect(flushes).toBe(1); - scheduler.schedule(); - expect(start).toHaveBeenCalledTimes(1); - tick(); - expect(flushes).toBe(2); - scheduler.cancel(); - expect(stop).toHaveBeenCalledWith(timer); - } finally { - start.mockRestore(); - stop.mockRestore(); - } - }); - - test("cancel drops pending data and allows restarting the clock", () => { - let flushes = 0; - let tick = () => {}; - const start = spyOn(globalThis, "setInterval").mockImplementation(((callback: () => void) => { - tick = callback as () => void; - return 123 as unknown as ReturnType; - }) as typeof setInterval); - const stop = spyOn(globalThis, "clearInterval").mockImplementation(() => {}); - try { - const scheduler = createLiveFlushScheduler(() => { flushes += 1; }); - scheduler.schedule(); - scheduler.cancel(); - scheduler.cancel(); - tick(); - expect(flushes).toBe(0); - expect(stop).toHaveBeenCalledTimes(1); - scheduler.schedule(); - expect(start).toHaveBeenCalledTimes(2); - tick(); - expect(flushes).toBe(1); - scheduler.cancel(); - } finally { - start.mockRestore(); - stop.mockRestore(); - } - }); -}); - describe("bootstrap refresh policy", () => { test("refreshes every tick while the live connection is down", () => { expect(shouldSyncBootstrap(false, 0)).toBe(true); diff --git a/frontend/src/refresh.ts b/frontend/src/refresh.ts index 2ef6d51..c95bd89 100644 --- a/frontend/src/refresh.ts +++ b/frontend/src/refresh.ts @@ -4,34 +4,6 @@ export const BOOTSTRAP_POLL_INTERVAL_MS = 15_000; export const BOOTSTRAP_LIVE_SYNC_INTERVAL_MS = 5 * 60_000; export const REMOTE_TASK_POLL_INTERVAL_MS = 2_000; export const REMOTE_TASK_POLL_TIMEOUT_MS = 60_000; -const LIVE_FLUSH_INTERVAL_MS = 1_000; - -export interface LiveFlushScheduler { - schedule: () => void; - cancel: () => void; -} - -export function createLiveFlushScheduler(flush: () => void, intervalMs = LIVE_FLUSH_INTERVAL_MS): LiveFlushScheduler { - let timer: ReturnType | undefined; - let pending = false; - return { - schedule: () => { - pending = true; - if (timer !== undefined) return; - timer = setInterval(() => { - if (!pending) return; - pending = false; - flush(); - }, intervalMs); - }, - cancel: () => { - pending = false; - if (timer === undefined) return; - clearInterval(timer); - timer = undefined; - }, - }; -} export function createRefreshQueue(execute: (quiet: boolean) => Promise) { let active: Promise | null = null; diff --git a/frontend/src/transport.test.ts b/frontend/src/transport.test.ts index 45ace10..0d1e867 100644 --- a/frontend/src/transport.test.ts +++ b/frontend/src/transport.test.ts @@ -1,6 +1,13 @@ import { describe, expect, spyOn, test } from "bun:test"; import { connectLive, reconnectDelay } from "./transport"; import type { BatchUpdate } from "./live"; +import { gzipSync } from "node:zlib"; + +function compressed(value: unknown): ArrayBuffer { + return Uint8Array.from(gzipSync(JSON.stringify(value))).buffer; +} + +async function settle() { await new Promise((resolve) => setTimeout(resolve, 20)); } describe("reconnectDelay", () => { test("backs off exponentially", () => { @@ -31,14 +38,15 @@ class FakeSocket { onopen: (() => void) | null = null; onclose: (() => void) | null = null; onerror: (() => void) | null = null; - onmessage: ((event: { data: string }) => void) | null = null; + binaryType = "blob"; + onmessage: ((event: { data: string | ArrayBuffer }) => void) | null = null; sent: string[] = []; failSend = false; constructor(public url: URL) { FakeSocket.instances.push(this); } open() { this.readyState = FakeSocket.OPEN; this.onopen?.(); } close() { this.readyState = FakeSocket.CLOSED; this.onclose?.(); } - receive(data: string) { this.onmessage?.({ data }); } + receive(data: string | ArrayBuffer) { this.onmessage?.({ data }); } send(data: string) { if (this.failSend) throw new Error("Disconnected"); this.sent.push(data); @@ -70,7 +78,6 @@ function transportHarness() { const dispose = connectLive({ serverId: "node/1" }, { onBatch: (batch) => { batches.push(batch); }, onConnectedChange: (connected) => { connections.push(connected); }, - onWakeRequested: async () => {}, }); return { sockets: FakeSocket.instances, @@ -172,7 +179,7 @@ describe("live connection lifecycle", () => { } finally { h.restore(); } }); - test("ignores callbacks from closed sockets and accepts new batches", () => { + test("ignores callbacks from closed sockets and decodes batches in wire order", async () => { const h = transportHarness(); try { const staleMessage = h.sockets[0].onmessage!; @@ -184,8 +191,40 @@ describe("live connection lifecycle", () => { expect(h.connections).toEqual([]); expect(h.batches).toEqual([]); h.sockets[1].open(); - h.sockets[1].receive('{"type":"batchUpdate","updates":[{"serverId":"new"}]}'); - expect(h.batches).toEqual([[{ serverId: "new" }]]); + expect(h.sockets[1].binaryType).toBe("arraybuffer"); + h.sockets[1].receive(compressed({ type: "batchUpdate", updates: [{ serverId: "first" }] })); + h.sockets[1].receive(compressed({ type: "batchUpdate", updates: [{ serverId: "second" }] })); + await settle(); + expect(h.batches).toEqual([[{ serverId: "first" }], [{ serverId: "second" }]]); + } finally { h.restore(); } + }); + + test("drops queued decoding after disconnect and reconnects on invalid gzip", async () => { + const h = transportHarness(); + try { + h.sockets[0].open(); + h.sockets[0].receive(compressed({ type: "batchUpdate", updates: [{ serverId: "stale" }] })); + h.visibility(true); + await settle(); + expect(h.batches).toEqual([]); + h.visibility(false); + h.sockets[1].open(); + h.sockets[1].receive(new Uint8Array([1, 2, 3]).buffer); + await settle(); + expect(h.sockets[1].readyState).toBe(FakeSocket.CLOSED); + h.advance(1_000); + expect(h.sockets).toHaveLength(3); + } finally { h.restore(); } + }); + + test("rejects decompression bombs", async () => { + const h = transportHarness(); + try { + h.sockets[0].open(); + h.sockets[0].receive(compressed("x".repeat(2 * 1024 * 1024 + 1))); + await settle(); + expect(h.sockets[0].readyState).toBe(FakeSocket.CLOSED); + expect(h.batches).toEqual([]); } finally { h.restore(); } }); }); diff --git a/frontend/src/transport.ts b/frontend/src/transport.ts index dd899d8..b809f8d 100644 --- a/frontend/src/transport.ts +++ b/frontend/src/transport.ts @@ -7,6 +7,30 @@ const STABLE_CONNECTION_MS = 10_000; const CONNECTION_TIMEOUT = 10_000; const HEARTBEAT_INTERVAL = 30_000; const HEARTBEAT_TIMEOUT = 10_000; +const MAX_MESSAGE_BYTES = 2 * 1024 * 1024; + +async function decodeMessage(data: ArrayBuffer) { + if (data.byteLength > MAX_MESSAGE_BYTES) throw new Error("Live message too large"); + const reader = new Blob([data]).stream().pipeThrough(new DecompressionStream("gzip")).getReader(); + const chunks: Uint8Array[] = []; + let size = 0; + try { + while (true) { + const { done, value } = await reader.read(); + if (done) break; + size += value.byteLength; + if (size > MAX_MESSAGE_BYTES) throw new Error("Expanded live message too large"); + chunks.push(value); + } + const bytes = new Uint8Array(size); + let offset = 0; + for (const chunk of chunks) { bytes.set(chunk, offset); offset += chunk.byteLength; } + return JSON.parse(new TextDecoder().decode(bytes)); + } finally { + await reader.cancel().catch(() => {}); + reader.releaseLock(); + } +} export function reconnectDelay(attempt: number, randomValue = Math.random()) { const exponent = Math.max(0, Math.min(30, Math.floor(attempt))); @@ -19,7 +43,6 @@ export function reconnectDelay(attempt: number, randomValue = Math.random()) { export interface LiveTransportHandlers { onBatch: (updates: BatchUpdate[]) => void; onConnectedChange: (connected: boolean) => void; - onWakeRequested: () => Promise; } export interface LiveTransportOptions { @@ -48,7 +71,6 @@ export function connectLive( let reconnectAttempt = 0; let connected = false; let openedAt: number | null = null; - let wakeInFlight = false; const setConnected = (next: boolean) => { if (connected === next) return; @@ -62,15 +84,6 @@ export function connectLive( reconnectTimer = null; }; - const wakeAgents = () => { - if (cancelled || suspended || wakeInFlight) return; - wakeInFlight = true; - void Promise.resolve() - .then(() => { if (!cancelled && !suspended) return handlers.onWakeRequested(); }) - .catch(() => {}) - .finally(() => { wakeInFlight = false; }); - }; - const scheduleReconnect = () => { if (cancelled || suspended || reconnectTimer !== null) return; const delay = reconnectDelay(reconnectAttempt); @@ -111,7 +124,6 @@ export function connectLive( deadlineTimer = window.setTimeout(() => closeCurrentSocket(true), HEARTBEAT_TIMEOUT); try { socket.send("ping"); - wakeAgents(); } catch { closeCurrentSocket(true); } @@ -128,6 +140,9 @@ export function connectLive( return; } socket = current; + current.binaryType = "arraybuffer"; + let messages = Promise.resolve(); + let queuedBytes = 0; deadlineTimer = window.setTimeout(() => closeCurrentSocket(true), CONNECTION_TIMEOUT); current.onopen = () => { if (socket !== current || cancelled || suspended) return; @@ -152,12 +167,20 @@ export function connectLive( heartbeatTimer = window.setTimeout(heartbeat, HEARTBEAT_INTERVAL); return; } - try { - const message = JSON.parse(event.data); - if (message.type === "batchUpdate" && Array.isArray(message.updates)) { + if (!(event.data instanceof ArrayBuffer)) return; + const data = event.data; + queuedBytes += data.byteLength; + if (queuedBytes > MAX_MESSAGE_BYTES * 4) { closeCurrentSocket(true); return; } + // Decompression is asynchronous; preserve wire order and drop stale results after reconnects. + messages = messages.then(async () => { + if (socket !== current) return; + const message = await decodeMessage(data); + if (socket === current && message.type === "batchUpdate" && Array.isArray(message.updates)) { handlers.onBatch(message.updates as BatchUpdate[]); } - } catch {} + }).catch(() => { + if (socket === current) closeCurrentSocket(true); + }).finally(() => { queuedBytes -= data.byteLength; }); }; }; diff --git a/scripts/agent-runtime-test.mjs b/scripts/agent-runtime-test.mjs index 592bfc4..e624d4a 100644 --- a/scripts/agent-runtime-test.mjs +++ b/scripts/agent-runtime-test.mjs @@ -9,6 +9,7 @@ import { tmpdir } from "node:os"; import { join, resolve } from "node:path"; import { setTimeout as delay } from "node:timers/promises"; import { WebSocketServer } from "ws"; +import { gunzipSync } from "node:zlib"; const binary = resolve(process.env.MONITOR_AGENT_BINARY ?? "agent/target/debug/nodeflare-agent"); const directory = mkdtempSync(join(tmpdir(), "nodeflare-agent-runtime-")); @@ -67,7 +68,7 @@ try { proxy.listen(0, "127.0.0.1"); await once(proxy, "listening"); - console.log("agent runtime: automatic update during one-second sampling"); + console.log("agent runtime: automatic update during three-second sampling"); websocketServer = new WebSocketServer({ noServer: true }); upgradeServer = createHttpServer(); let redirected = false; @@ -89,27 +90,30 @@ try { socket.send(JSON.stringify({ type: "config", ts: Math.floor(Date.now() / 1_000), config: { - report_interval: 15, collect_interval: 1, network_interface: "", + report_interval: 15, collect_interval: 3, network_interface: "", agent_mirror: "", auto_update: true, latency_tasks: [], }, })); - socket.on("message", (raw) => { - const message = JSON.parse(raw.toString()); - if (message.type !== "update") return; + let receivedThrough = 0; + socket.on("message", (raw, binary) => { + if (!binary) return; + const message = JSON.parse(gunzipSync(raw).toString()); samples += message.samples.length; - if (message.persist) persistedThrough = Math.max(...message.samples.map((sample) => sample.timestamp)); + receivedThrough = Math.max(receivedThrough, ...message.samples.map((sample) => sample.metrics.timestamp)); + if (!message.persist) return; + persistedThrough = receivedThrough; socket.send(JSON.stringify({ type: "ack", ts: Math.floor(Date.now() / 1_000), persisted: message.persist, persistenceError: false, persistedThroughTs: persistedThrough, - nextPersistAfterMs: 1_000, nextWssReportAfterMs: 1_000, realtimeHint: false, + nextPersistAfterMs: 15_000, })); }); }); let token; for (let index = 0; index < 100_000; index++) { const candidate = `runtime-update-${index}`; - if (createHash("sha256").update(candidate).digest().readBigUInt64BE(0) % 1_801n === 2n) { + if (createHash("sha256").update(candidate).digest().readBigUInt64BE(0) % 1_801n === 5n) { token = candidate; break; } @@ -121,6 +125,40 @@ try { assert(redirected, "Same-origin WebSocket redirects must remain supported"); await stopAgent(updating); + console.log("agent runtime: three-second uploads are unique and reconnect replays unconfirmed samples"); + const telemetryConnections = []; + const trackTelemetry = (socket) => { + const connection = { timestamps: [], durable: 0, firstInfo: false, infoCount: 0, commits: 0 }; + telemetryConnections.push(connection); + socket.on("message", (raw, binary) => { + if (!binary) return; + const message = JSON.parse(gunzipSync(raw).toString()); + for (const sample of message.samples) { + if (!connection.timestamps.length) connection.firstInfo = Boolean(sample.info); + connection.infoCount += Number(Boolean(sample.info)); + assert(!connection.timestamps.includes(sample.metrics.timestamp), "Sample was uploaded twice on the same connection"); + connection.timestamps.push(sample.metrics.timestamp); + } + if (message.persist) { + connection.durable = connection.timestamps.at(-1); + connection.commits++; + } + if (telemetryConnections.length === 1 && connection.timestamps.length === 3) socket.terminate(); + }); + }; + websocketServer.on("connection", trackTelemetry); + const telemetry = startAgent(upgradeServer.address().port, "runtime-telemetry", "telemetry"); + await waitUntil(() => telemetryConnections.length >= 2 && telemetryConnections[1].commits >= 2, + 40_000, "Telemetry reconnect or commit did not complete"); + await stopAgent(telemetry); + websocketServer.off("connection", trackTelemetry); + const [before, after] = telemetryConnections; + assert(before.firstInfo && after.firstInfo, "Static info must be resent on reconnect"); + assert.equal(before.infoCount, 1, "Unchanged static info must not be uploaded repeatedly"); + assert(before.timestamps.slice(1).every((timestamp) => after.timestamps.includes(timestamp)), "Unconfirmed samples were lost"); + assert(!after.timestamps.includes(before.durable), "Durable samples must not be replayed"); + assert(before.timestamps.slice(1).every((timestamp, index) => timestamp - before.timestamps[index] >= 3), "Sampling interval is shorter than three seconds"); + console.log("agent runtime: remote command survives disconnect and returns its result"); const taskId = randomUUID(); let remoteConnections = 0; @@ -128,7 +166,8 @@ try { let remoteResult; websocketServer.on("connection", (socket) => { remoteConnections++; - socket.on("message", (raw) => { + socket.on("message", (raw, binary) => { + if (binary) return; const message = JSON.parse(raw.toString()); if (message.type === "task_received" && message.task_id === taskId) { received = true; diff --git a/scripts/smoke-test.sh b/scripts/smoke-test.sh index d89cdb6..4c102c9 100644 --- a/scripts/smoke-test.sh +++ b/scripts/smoke-test.sh @@ -272,7 +272,7 @@ MONITOR_BASE_URL="$MONITOR_BASE_URL" MONITOR_ADMIN_TOKEN="$admin_token" \ if [ -n "${MONITOR_AGENT_BINARY:-}" ]; then step "real Agent failed-latency and persistence" request -H "Authorization: Bearer $admin_token" -H 'Content-Type: application/json' -X PATCH \ - --data "$(printf '%s' "$server_input" | jq '.report_interval=15 | .collect_interval=1')" \ + --data "$(printf '%s' "$server_input" | jq '.report_interval=15 | .collect_interval=3')" \ "$MONITOR_BASE_URL/api/admin/servers/$server_id" >/dev/null agent_latency_task_id=$(request -H "Authorization: Bearer $admin_token" \ -H 'Content-Type: application/json' \ diff --git a/scripts/websocket-smoke.mjs b/scripts/websocket-smoke.mjs index b8e7e3c..c179a91 100644 --- a/scripts/websocket-smoke.mjs +++ b/scripts/websocket-smoke.mjs @@ -1,6 +1,22 @@ import WebSocket from "ws"; import assert from "node:assert/strict"; import { once } from "node:events"; +import { gzipSync, gunzipSync } from "node:zlib"; + +function telemetryPayload(reports, persist) { + return gzipSync(JSON.stringify({ + samples: reports.map((report) => { + const metrics = { ...report }; + const info = {}; + for (const key of ["cpu_model", "os", "kernel", "arch", "virtualization", "gpu_model", "agent_version", "ip_v4", "ip_v6"]) { + info[key] = metrics[key] ?? ""; + delete metrics[key]; + } + return { metrics, info }; + }), + persist, + })); +} const baseUrl = process.env.MONITOR_BASE_URL; const adminToken = process.env.MONITOR_ADMIN_TOKEN; @@ -190,10 +206,10 @@ function waitForJsonMessage(socket, expected, predicate) { cleanup(); reject(error); }; - const onMessage = (data) => { + const onMessage = (data, binary) => { let message; try { - message = JSON.parse(data.toString()); + message = JSON.parse((binary ? gunzipSync(data) : data).toString()); } catch { return; } @@ -358,7 +374,7 @@ try { name: "Offline remote smoke", region: "", group_name: "", tags: "", hidden: true, expires_at: null, traffic_limit: 0, traffic_limit_type: "max", price: 0, billing_cycle: 30, currency: "USD", auto_renewal: false, network_interface: "", - reset_day: 1, report_interval: 60, collect_interval: 1, rx_correction: 0, + reset_day: 1, report_interval: 60, collect_interval: 3, rx_correction: 0, tx_correction: 0, agent_mirror: "", offline_notify_disabled: true, auto_update: false, }), }); @@ -554,27 +570,6 @@ try { }); } - const wakeHintPromise = waitForJsonMessage( - agent, - "batched overview wake hint", - (message) => message.type === "ack" && message.realtimeHint === true, - ); - const wakeResponse = await fetch(new URL("/api/live/wake", baseUrl), { - method: "POST", - headers: { - Authorization: `Bearer ${adminToken}`, - "Content-Type": "application/json", - }, - body: JSON.stringify({ server_ids: [serverId] }), - }); - if (wakeResponse.status !== 204) { - throw new Error(`Batch wake returned HTTP ${wakeResponse.status}: ${await wakeResponse.text()}`); - } - const wakeHint = await wakeHintPromise; - if (wakeHint.nextWssReportAfterMs !== 5_000) { - throw new Error(`Invalid batch wake hint: ${JSON.stringify(wakeHint)}`); - } - const timestamp = Math.floor(Date.now() / 1_000); const baseSample = { timestamp, @@ -654,22 +649,10 @@ try { sample(5, { net_rx_total: 536870912, net_tx_total: 268435456 }), ]; const latestTimestamp = timestamp + 5; - for (const persist of [undefined, null]) { - const rejectedAck = waitForJsonMessage( - agent, - "invalid persistence flag ACK", - (message) => message.type === "ack" && message.realtimeHint === false, - ); - agent.send(JSON.stringify({ type: "update", batchId: "invalid-persist", samples, persist })); - const rejected = await rejectedAck; - if (rejected.persistenceError !== true || rejected.persisted !== false) { - throw new Error(`Accepted invalid persistence flag: ${JSON.stringify(rejected)}`); - } - } const ackPromise = waitForJsonMessage( agent, "Agent metric ACK", - (message) => message.type === "ack" && message.realtimeHint === false, + (message) => message.type === "ack", ); const updatePromise = waitForJsonMessage( dashboard, @@ -684,7 +667,7 @@ try { ), ), ); - agent.send(JSON.stringify({ type: "update", batchId: `smoke-${timestamp}`, samples, persist: true })); + agent.send(telemetryPayload(samples, true)); const [ack, update] = await Promise.all([ackPromise, updatePromise]); if ( ack.ts <= 0 || @@ -692,11 +675,12 @@ try { ack.persistenceError !== false || !Number.isInteger(ack.persistedThroughTs) || ack.persistedThroughTs < latestTimestamp || - ack.nextPersistAfterMs !== 60_000 || - ack.nextWssReportAfterMs !== 5_000 + ack.nextPersistAfterMs !== 60_000 ) { throw new Error(`Invalid Agent metric ACK: ${JSON.stringify(ack)}`); } + assert.equal(update.updates[0].samples.length, 1, "Browser receives only the latest metric snapshot"); + assert.equal(update.updates[0].samples[0].data.latency_results.length, 2, "All new latency results are retained"); const publicIdentityFields = [ "cpu_model", diff --git a/shared/Cargo.lock b/shared/Cargo.lock new file mode 100644 index 0000000..0fbc01d --- /dev/null +++ b/shared/Cargo.lock @@ -0,0 +1,162 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 4 + +[[package]] +name = "adler2" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa" + +[[package]] +name = "cfg-if" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" + +[[package]] +name = "crc32fast" +version = "1.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8498c871161e1742aaa9d52551b2d6ebdd4c3d45a3be423e3728f33b955be550" +dependencies = [ + "cfg-if", +] + +[[package]] +name = "flate2" +version = "1.1.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6e634e2e0ebac1ee034020da1ca582e17ffe4e0f5e985823721e168928136dcb" +dependencies = [ + "crc32fast", + "miniz_oxide", + "zlib-rs", +] + +[[package]] +name = "itoa" +version = "1.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" + +[[package]] +name = "memchr" +version = "2.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" + +[[package]] +name = "miniz_oxide" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b63fbc4a50860e98e7b2aa7804ded1db5cbc3aff9193adaff57a6931bf7c4b4c" +dependencies = [ + "adler2", + "simd-adler32", +] + +[[package]] +name = "nodeflare-telemetry" +version = "1.0.0" +dependencies = [ + "flate2", + "serde", + "serde_json", +] + +[[package]] +name = "proc-macro2" +version = "1.0.107" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "985e7ec9bb745e6ce6535b544d84d6cd6f7ad8bd711c398938ae983b91a766d9" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "quote" +version = "1.0.47" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fbf4db142a473a8d80c26bbf18454ed458bf8d26c8219c331daecfdbd079001" +dependencies = [ + "proc-macro2", +] + +[[package]] +name = "serde" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba" +dependencies = [ + "serde_core", + "serde_derive", +] + +[[package]] +name = "serde_core" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "67dca2c9c51e58a4791a4b1ed58308b39c64224d349a935ab5039aa360942a48" +dependencies = [ + "serde_derive", +] + +[[package]] +name = "serde_derive" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "serde_json" +version = "1.0.151" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c841b55ecdae098c80dcae9cf767f6f8a0c2cdb3416bbef72181df4d0fe73f14" +dependencies = [ + "itoa", + "memchr", + "serde", + "serde_core", + "zmij", +] + +[[package]] +name = "simd-adler32" +version = "0.3.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3a219298ac11a56ea9a6d2120044824d6f01aeb034955e7af7bc16858527deea" + +[[package]] +name = "syn" +version = "3.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "12df2e0110f65b775f769bb17ef989067a1d931b2eb822bd4346631eeada89f9" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "unicode-ident" +version = "1.0.24" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" + +[[package]] +name = "zlib-rs" +version = "0.6.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "34b31d188d9d685a4f9c7b46d6e36631b07058d2cfe190267adce54dc230bf12" + +[[package]] +name = "zmij" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b" diff --git a/shared/Cargo.toml b/shared/Cargo.toml new file mode 100644 index 0000000..1e5a180 --- /dev/null +++ b/shared/Cargo.toml @@ -0,0 +1,14 @@ +[package] +name = "nodeflare-telemetry" +version = "1.0.0" +edition = "2024" +license = "MIT" +publish = false + +[lib] +path = "telemetry.rs" + +[dependencies] +flate2 = "1" +serde = { version = "1", features = ["derive"] } +serde_json = "1" diff --git a/shared/telemetry.rs b/shared/telemetry.rs new file mode 100644 index 0000000..f45d8f3 --- /dev/null +++ b/shared/telemetry.rs @@ -0,0 +1,347 @@ +use flate2::{Compression, read::GzDecoder, write::GzEncoder}; +use serde::{Deserialize, Serialize, de::DeserializeOwned}; +use std::io::{self, Read}; + +pub const MIN_COLLECT_INTERVAL: u64 = 3; +pub const MAX_BATCH_BYTES: usize = 768 * 1024; +pub const MAX_DECODED_BYTES: usize = 2 * 1024 * 1024; +pub const MAX_SAMPLES: usize = 720; + +#[derive(Debug, Clone, Default, Deserialize, Serialize)] +#[serde(default, deny_unknown_fields)] +pub struct Report { + pub timestamp: i64, + pub cpu: f64, + pub load1: f64, + pub load5: f64, + pub load15: f64, + pub mem_used: i64, + pub mem_total: i64, + pub swap_used: i64, + pub swap_total: i64, + pub disk_used: i64, + pub disk_total: i64, + pub net_in: f64, + pub net_out: f64, + pub net_rx_total: i64, + pub net_tx_total: i64, + pub uptime: i64, + pub processes: i64, + pub tcp_connections: i64, + pub udp_connections: i64, + pub cpu_cores: i64, + #[serde(skip_serializing_if = "String::is_empty")] + pub cpu_model: String, + #[serde(skip_serializing_if = "String::is_empty")] + pub os: String, + #[serde(skip_serializing_if = "String::is_empty")] + pub kernel: String, + #[serde(skip_serializing_if = "String::is_empty")] + pub arch: String, + #[serde(skip_serializing_if = "String::is_empty")] + pub virtualization: String, + pub gpu_usage: f64, + #[serde(skip_serializing_if = "String::is_empty")] + pub gpu_model: String, + #[serde(skip_serializing_if = "String::is_empty")] + pub agent_version: String, + #[serde(skip_serializing_if = "String::is_empty")] + pub ip_v4: String, + #[serde(skip_serializing_if = "String::is_empty")] + pub ip_v6: String, + pub disk_read_bps: f64, + pub disk_write_bps: f64, + pub disk_read_iops: f64, + pub disk_write_iops: f64, + pub disk_await_ms: f64, + pub disk_utilization: f64, + pub disks: Vec, + pub gpus: Vec, + pub latency_results: Vec, +} + +impl Report { + fn round_for_upload(&mut self) { + for value in [ + &mut self.net_in, + &mut self.net_out, + &mut self.disk_read_bps, + &mut self.disk_write_bps, + ] { + *value = value.round(); + } + for value in [ + &mut self.cpu, + &mut self.load1, + &mut self.load5, + &mut self.load15, + &mut self.gpu_usage, + &mut self.disk_read_iops, + &mut self.disk_write_iops, + &mut self.disk_await_ms, + &mut self.disk_utilization, + ] { + *value = (*value * 100.0).round() / 100.0; + } + for disk in &mut self.disks { + disk.read_bps = disk.read_bps.round(); + disk.write_bps = disk.write_bps.round(); + for value in [ + &mut disk.read_iops, + &mut disk.write_iops, + &mut disk.await_ms, + &mut disk.utilization, + ] { + *value = (*value * 100.0).round() / 100.0; + } + } + for gpu in &mut self.gpus { + gpu.usage = gpu.usage.map(|value| (value * 100.0).round() / 100.0); + } + for latency in &mut self.latency_results { + latency.latency_ms = (latency.latency_ms * 100.0).round() / 100.0; + latency.packet_loss = (latency.packet_loss * 100.0).round() / 100.0; + } + } +} + +#[derive(Debug, Clone, Default, Deserialize, Serialize)] +#[serde(default, deny_unknown_fields)] +pub struct DiskMetric { + pub name: String, + pub mount_point: String, + pub used: i64, + pub total: i64, + pub read_bps: f64, + pub write_bps: f64, + pub read_iops: f64, + pub write_iops: f64, + pub await_ms: f64, + pub utilization: f64, +} + +#[derive(Debug, Clone, Default, Deserialize, Serialize)] +#[serde(default, deny_unknown_fields)] +pub struct GpuMetric { + pub model: String, + pub usage: Option, + pub memory_used: i64, + pub memory_total: i64, +} + +#[derive(Debug, Clone, Default, Deserialize, Serialize)] +#[serde(default, deny_unknown_fields)] +pub struct LatencyResult { + pub task_id: String, + pub timestamp: i64, + pub latency_ms: f64, + pub packet_loss: f64, +} + +#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)] +#[serde(deny_unknown_fields)] +pub struct Info { + cpu_model: String, + os: String, + kernel: String, + arch: String, + virtualization: String, + gpu_model: String, + agent_version: String, + ip_v4: String, + ip_v6: String, +} + +impl Info { + fn take(report: &mut Report) -> Self { + Self { + cpu_model: std::mem::take(&mut report.cpu_model), + os: std::mem::take(&mut report.os), + kernel: std::mem::take(&mut report.kernel), + arch: std::mem::take(&mut report.arch), + virtualization: std::mem::take(&mut report.virtualization), + gpu_model: std::mem::take(&mut report.gpu_model), + agent_version: std::mem::take(&mut report.agent_version), + ip_v4: std::mem::take(&mut report.ip_v4), + ip_v6: std::mem::take(&mut report.ip_v6), + } + } + + fn apply(&self, report: &mut Report) { + report.cpu_model.clone_from(&self.cpu_model); + report.os.clone_from(&self.os); + report.kernel.clone_from(&self.kernel); + report.arch.clone_from(&self.arch); + report.virtualization.clone_from(&self.virtualization); + report.gpu_model.clone_from(&self.gpu_model); + report.agent_version.clone_from(&self.agent_version); + report.ip_v4.clone_from(&self.ip_v4); + report.ip_v6.clone_from(&self.ip_v6); + } +} + +#[derive(Debug, Deserialize, Serialize)] +#[serde(deny_unknown_fields)] +pub struct Sample { + pub metrics: Report, + #[serde(skip_serializing_if = "Option::is_none")] + pub info: Option, +} + +#[derive(Debug, Deserialize, Serialize)] +#[serde(deny_unknown_fields)] +pub struct Update { + pub samples: Vec, + pub persist: bool, +} + +impl Update { + pub fn from_reports(reports: Vec, persist: bool, info: &mut Option) -> Self { + let samples = reports + .into_iter() + .map(|mut metrics| { + let current = Info::take(&mut metrics); + metrics.round_for_upload(); + let changed = info.as_ref() != Some(¤t); + if changed { + *info = Some(current.clone()); + } + Sample { + metrics, + info: changed.then_some(current), + } + }) + .collect(); + Self { samples, persist } + } + + pub fn into_reports(self, info: &mut Option) -> io::Result> { + if self.samples.len() > MAX_SAMPLES { + return Err(io::Error::other("too many telemetry samples")); + } + self.samples + .into_iter() + .map(|mut sample| { + if let Some(current) = sample.info { + *info = Some(current); + } + info.as_ref() + .ok_or_else(|| io::Error::other("missing telemetry info"))? + .apply(&mut sample.metrics); + Ok(sample.metrics) + }) + .collect() + } +} + +pub fn encode(value: &impl Serialize) -> io::Result> { + let mut encoder = GzEncoder::new(Vec::new(), Compression::fast()); + serde_json::to_writer(&mut encoder, value)?; + encoder.finish() +} + +pub fn decode(bytes: &[u8]) -> io::Result { + if bytes.len() > MAX_DECODED_BYTES { + return Err(io::Error::other("telemetry frame is too large")); + } + let mut decoded = Vec::new(); + GzDecoder::new(bytes) + .take(MAX_DECODED_BYTES as u64 + 1) + .read_to_end(&mut decoded)?; + if decoded.len() > MAX_DECODED_BYTES { + return Err(io::Error::other("expanded telemetry frame is too large")); + } + serde_json::from_slice(&decoded).map_err(io::Error::other) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn trims_float_noise_without_changing_byte_counters() { + let report = Report { + cpu: 100.0 / 3.0, + net_in: 12345.6789, + net_rx_total: i64::MAX, + disks: vec![DiskMetric { + read_bps: 125.6789, + await_ms: 1.234567, + ..DiskMetric::default() + }], + latency_results: vec![LatencyResult { + latency_ms: 28.4444, + packet_loss: 100.0 / 3.0, + ..LatencyResult::default() + }], + ..Report::default() + }; + let encoded = encode(&Update::from_reports(vec![report], false, &mut None)).unwrap(); + let reports = decode::(&encoded) + .unwrap() + .into_reports(&mut None) + .unwrap(); + assert_eq!(reports[0].cpu, 33.33); + assert_eq!(reports[0].net_in, 12346.0); + assert_eq!(reports[0].net_rx_total, i64::MAX); + assert_eq!(reports[0].disks[0].read_bps, 126.0); + assert_eq!(reports[0].disks[0].await_ms, 1.23); + assert_eq!(reports[0].latency_results[0].latency_ms, 28.44); + } + + #[test] + fn info_is_sent_on_connect_and_changes_including_during_replay() { + let mut encoder = None; + let mut decoder = None; + for (name, changed) in [ + ("first", true), + ("first", false), + ("second", true), + ("first", true), + ] { + let update = Update::from_reports( + vec![Report { + cpu_model: name.into(), + cpu: 23.5, + ..Report::default() + }], + false, + &mut encoder, + ); + assert_eq!(update.samples[0].info.is_some(), changed); + assert!(update.samples[0].metrics.cpu_model.is_empty()); + let bytes = encode(&update).unwrap(); + let decoded: Update = decode(&bytes).unwrap(); + let reports = decoded.into_reports(&mut decoder).unwrap(); + assert_eq!(reports[0].cpu_model, name); + assert_eq!(reports[0].cpu, 23.5); + } + let update = Update::from_reports(vec![Report::default()], true, &mut None); + assert!(update.samples[0].info.is_some()); + } + + #[test] + fn rejects_missing_info_corruption_and_decompression_bombs() { + let update = Update { + samples: vec![Sample { + metrics: Report::default(), + info: None, + }], + persist: false, + }; + assert!(update.into_reports(&mut None).is_err()); + assert!(decode::(b"not gzip").is_err()); + let bytes = encode(&"x".repeat(MAX_DECODED_BYTES + 1)).unwrap(); + assert!(decode::(&bytes).is_err()); + } + + #[test] + fn compresses_batches_and_supports_empty_commit_messages() { + let update = Update::from_reports(vec![Report::default(); 20], false, &mut None); + assert!(encode(&update).unwrap().len() < serde_json::to_vec(&update).unwrap().len() / 4); + let update = Update::from_reports(Vec::new(), true, &mut None); + let decoded: Update = decode(&encode(&update).unwrap()).unwrap(); + assert!(decoded.persist); + assert!(decoded.samples.is_empty()); + } +}