diff --git a/README.md b/README.md index 015c7a7..c1449f7 100644 --- a/README.md +++ b/README.md @@ -2,6 +2,8 @@ NodeFlare 是一款轻量级的自托管服务器监控面板，通过 Web 界面查看服务器状态，并由轻量级 Agent 采集数据上报。支持实时监控、TCP 拨测、Telegram 告警、远程执行、TOTP 两步验证、主题定制、SQLite/PostgreSQL 备份迁移等。 +CPU、内存和网速每秒采样，默认每 3 秒压缩批量上传。服务器卡片每秒显示一个真实采样值，正常连接下有约 2～3 秒的显示延迟；断线重连后跳过积压的旧数据。磁盘容量、GPU 等较慢指标单独缓存，历史数据按节点设置的保存间隔聚合。增大实时上传间隔会降低卡片连续更新的频率。 + ## 安装服务端 支持 Linux x64/ARM64、Windows x64、macOS ARM64、FreeBSD 13+ x64/ARM64。安装脚本自动下载最新 Release 并注册系统服务。 diff --git a/agent/src/main.rs b/agent/src/main.rs index 47d0ca0..59fd659 100644 --- a/agent/src/main.rs +++ b/agent/src/main.rs @@ -67,6 +67,8 @@ const LIVE_ACK_READ_TIMEOUT: Duration = Duration::from_secs(5); const LIVE_HINT_READ_TIMEOUT: Duration = Duration::from_millis(10); const BASIC_INFO_REFRESH_INTERVAL: Duration = Duration::from_secs(5 * 60); const SLOW_METRICS_REFRESH_INTERVAL: Duration = Duration::from_secs(5); +const SAMPLE_INTERVAL: Duration = Duration::from_secs(1); +const GPU_METRICS_REFRESH_INTERVAL: Duration = Duration::from_secs(5); const PUBLIC_IP_REFRESH_INTERVAL: Duration = Duration::from_secs(10 * 60); const PUBLIC_IP_STALE_AFTER: Duration = Duration::from_secs(30 * 60); const PUBLIC_IP_PROBE_TIMEOUT: Duration = Duration::from_secs(5); @@ -87,6 +89,7 @@ struct RuntimeConfig { token: String, endpoint: String, report_interval: u64, + // Wire/config field name; controls upload batching, not the one-second sampler. collect_interval: u64, network_interface: String, agent_mirror: String, @@ -445,7 +448,7 @@ impl LiveSender { } fn live_batch_interval(collect_interval: u64) -> Duration { - Duration::from_secs(collect_interval.clamp(telemetry::MIN_COLLECT_INTERVAL, 60)) + Duration::from_secs(collect_interval.clamp(telemetry::MIN_UPLOAD_INTERVAL, 60)) } #[derive(Debug, Default, Clone)] @@ -520,11 +523,67 @@ fn cpu_sample() -> CpuSample { } fn selected_interface(name: &str, filter: &str) -> bool { - if filter.trim().is_empty() { - name != "lo" && !name.to_ascii_lowercase().contains("loopback") - } else { - filter.split(',').any(|value| value.trim() == name) + let name = name.trim(); + if name.is_empty() { + return false; } + let mut has_include = false; + let mut included = false; + for pattern in filter + .split(',') + .map(str::trim) + .filter(|value| !value.is_empty()) + { + if let Some(pattern) = pattern + .strip_prefix('!') + .or_else(|| pattern.strip_prefix('-')) + { + if !pattern.is_empty() && wildcard_match(pattern, name) { + return false; + } + } else { + has_include = true; + included |= wildcard_match(pattern, name); + } + } + // Explicit includes can select a bridge; automatic selection avoids double counting. + if has_include { + return included; + } + let lower = name.to_ascii_lowercase(); + !lower.contains("loopback") + && ![ + "lo", "br", "cni", "docker", "podman", "flannel", "veth", "virbr", "vmbr", "tap", + "fwbr", "fwpr", + ] + .iter() + .any(|prefix| lower.starts_with(prefix)) +} + +fn wildcard_match(pattern: &str, value: &str) -> bool { + let pattern = pattern.as_bytes(); + let value = value.as_bytes(); + let (mut p, mut v, mut star, mut star_value) = (0, 0, None, 0); + while v < value.len() { + if p < pattern.len() && (pattern[p] == b'?' || pattern[p] == value[v]) { + p += 1; + v += 1; + } else if p < pattern.len() && pattern[p] == b'*' { + star = Some(p); + star_value = v; + p += 1; + } else if let Some(star_position) = star { + p = star_position + 1; + star_value += 1; + v = star_value; + } else { + return false; + } + } + while p < pattern.len() && pattern[p] == b'*' { + p += 1; + } + p == pattern.len() } #[cfg(target_os = "linux")] @@ -638,35 +697,95 @@ fn connection_counts() -> (i64, i64) { fn disk_usage() -> Vec { let output = command( "df", - &[ - "-B1", - "-l", - "--output=source,target,size,used", - "-x", - "tmpfs", - "-x", - "devtmpfs", - ], + &["-B1", "-l", "--output=source,target,fstype,size,used"], ); output .lines() .skip(1) .filter_map(|line| { let fields = line.split_whitespace().collect::>(); - if fields.len() < 4 { + if fields.len() < 5 { return None; } - let total = fields[2].parse::().ok()?; - let used = fields[3].parse::().ok()?; + let source = fields[0]; + let mount_point = fields[1..fields.len() - 3].join(" "); + let filesystem = fields[fields.len() - 3]; + if excluded_filesystem(filesystem, &mount_point) { + return None; + } + let total = fields[fields.len() - 2].parse::().ok()?; + let used = fields[fields.len() - 1].parse::().ok()?; (total > 0).then(|| DiskMetric { - name: fields[0].to_string(), - mount_point: fields[1].to_string(), + name: source.to_string(), + mount_point, used, total, ..DiskMetric::default() }) }) - .collect() + .collect::>() +} + +#[cfg(target_os = "linux")] +fn excluded_filesystem(filesystem: &str, mount_point: &str) -> bool { + let mount_point = mount_point.to_ascii_lowercase(); + // Keep the root filesystem visible even when a container reports it as overlay. + if mount_point == "/" { + return false; + } + let filesystem = filesystem.to_ascii_lowercase(); + [ + "tmpfs", + "devtmpfs", + "devpts", + "proc", + "sysfs", + "cgroup", + "cgroup2", + "overlay", + "squashfs", + "efivarfs", + "pstore", + "mqueue", + "hugetlbfs", + "debugfs", + "fusectl", + ] + .iter() + .any(|value| filesystem == *value || filesystem.starts_with(&format!("{value}."))) + || ["/proc", "/sys", "/run", "/dev"] + .iter() + .any(|prefix| mount_point == *prefix || mount_point.starts_with(&format!("{prefix}/"))) +} + +#[cfg(any(target_os = "linux", test))] +fn disk_identity(name: &str) -> String { + let name = name.trim(); + // ZFS reports datasets as pool/dataset. Keep ordinary absolute paths and + // remote sources separate because their slash is part of the path. + if !name.starts_with('/') && !name.contains(':') && name.contains('/') { + return name.split('/').next().unwrap_or(name).to_string(); + } + name.to_string() +} + +#[cfg(any(target_os = "linux", test))] +fn dedupe_disks(disks: impl IntoIterator) -> Vec { + let mut unique = HashMap::::new(); + for disk in disks { + let key = disk_identity(&disk.name); + let replace = unique.get(&key).is_none_or(|current| { + disk.total > current.total + || (disk.total == current.total + && disk.mount_point.len() < current.mount_point.len()) + }); + if replace { + unique.insert(key, disk); + } + } + let mut disks = unique.into_values().collect::>(); + disks.sort_by(|left, right| left.mount_point.cmp(&right.mount_point)); + disks } #[cfg(target_os = "linux")] @@ -1238,6 +1357,7 @@ struct Collector { network_interface: String, basic: BasicMetrics, basic_at: Instant, + gpu_at: Instant, slow: SlowMetrics, slow_at: Instant, public_ip: PublicIpProbe, @@ -1253,6 +1373,7 @@ impl Collector { network_interface: config.network_interface.clone(), basic: BasicMetrics::default(), basic_at: Instant::now(), + gpu_at: Instant::now(), slow: SlowMetrics::default(), slow_at: Instant::now(), public_ip: PublicIpProbe::default(), @@ -1280,6 +1401,7 @@ impl Collector { gpus, }; self.basic_at = Instant::now(); + self.gpu_at = Instant::now(); } fn refresh_slow(&mut self) { @@ -1295,7 +1417,7 @@ impl Collector { .count() as i64 }); self.slow = SlowMetrics { - disks: disk_usage(), + disks: dedupe_disks(disk_usage()), processes, tcp_connections: file_line_count("/proc/net/tcp") + file_line_count("/proc/net/tcp6"), udp_connections: file_line_count("/proc/net/udp") + file_line_count("/proc/net/udp6"), @@ -1337,6 +1459,9 @@ impl Collector { if self.basic_at.elapsed() >= BASIC_INFO_REFRESH_INTERVAL { self.refresh_basic(); } + if self.gpu_at.elapsed() >= GPU_METRICS_REFRESH_INTERVAL { + self.refresh_gpu_metrics(); + } if self.slow_at.elapsed() >= SLOW_METRICS_REFRESH_INTERVAL { self.refresh_slow(); } @@ -1445,6 +1570,7 @@ struct Collector { previous_at: Instant, basic: BasicMetrics, basic_at: Instant, + gpu_at: Instant, slow: SlowMetrics, slow_at: Instant, public_ip: PublicIpProbe, @@ -1460,6 +1586,7 @@ impl Collector { previous_at: Instant::now(), basic: BasicMetrics::default(), basic_at: Instant::now(), + gpu_at: Instant::now(), slow: SlowMetrics::default(), slow_at: Instant::now(), public_ip: PublicIpProbe::default(), @@ -1492,6 +1619,7 @@ impl Collector { gpus, }; self.basic_at = Instant::now(); + self.gpu_at = Instant::now(); } fn refresh_slow(&mut self) { @@ -1552,6 +1680,9 @@ impl Collector { if self.basic_at.elapsed() >= BASIC_INFO_REFRESH_INTERVAL { self.refresh_basic(); } + if self.gpu_at.elapsed() >= GPU_METRICS_REFRESH_INTERVAL { + self.refresh_gpu_metrics(); + } if self.slow_at.elapsed() >= SLOW_METRICS_REFRESH_INTERVAL { self.refresh_slow(); } @@ -1619,6 +1750,32 @@ impl Collector { } } +impl Collector { + fn refresh_gpu_metrics(&mut self) { + self.gpu_at = Instant::now(); + // Unsupported devices are rediscovered with basic info, not every sample. + if !self + .basic + .gpus + .iter() + .any(|gpu| gpu.usage.is_some() || gpu.memory_total > 0) + { + return; + } + let detailed = nvidia_gpu_info(); + if detailed.is_empty() { + return; + } + self.basic.gpu_usage = average_gpu_usage(&detailed); + self.basic.gpu_model = detailed + .iter() + .map(|gpu| gpu.model.as_str()) + .collect::>() + .join(" · "); + self.basic.gpus = detailed; + } +} + fn runtime_config(options: &CliOptions) -> Result { let interval = options.interval; if !(15..=3600).contains(&interval) { @@ -1642,7 +1799,7 @@ fn runtime_config(options: &CliOptions) -> Result { token, endpoint: endpoint.trim_end_matches('/').to_string(), report_interval: interval, - collect_interval: telemetry::MIN_COLLECT_INTERVAL, + collect_interval: telemetry::MIN_UPLOAD_INTERVAL, network_interface: String::new(), agent_mirror: String::new(), auto_update: true, @@ -2579,7 +2736,7 @@ fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) { } = worker; let mut socket: Option = None; let mut wss_interval = configured_interval.lock().map_or( - Duration::from_secs(telemetry::MIN_COLLECT_INTERVAL), + Duration::from_secs(telemetry::MIN_UPLOAD_INTERVAL), |interval| *interval, ); let mut info = None; @@ -2720,7 +2877,7 @@ fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) { if let Ok(interval) = configured_interval.lock() { wss_interval = (*interval).clamp( - Duration::from_secs(telemetry::MIN_COLLECT_INTERVAL), + Duration::from_secs(telemetry::MIN_UPLOAD_INTERVAL), Duration::from_secs(60), ); } @@ -2806,7 +2963,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.max(telemetry::MIN_COLLECT_INTERVAL); + config.collect_interval = remote.collect_interval.max(telemetry::MIN_UPLOAD_INTERVAL); } config .network_interface @@ -3242,7 +3399,7 @@ fn run(options: &CliOptions, once: bool, print_only: bool) -> Result<()> { if once || print_only { collector.public_ip.wait_initial(); } - let mut next_collect = Instant::now() + Duration::from_secs(config.collect_interval); + let mut next_collect = Instant::now() + SAMPLE_INTERVAL; let mut next_latency: HashMap = HashMap::new(); let stats = Arc::new(runtime_stats::RuntimeStats::default()); let mut latency_executor = LatencyExecutor::new(Arc::clone(&stats))?; @@ -3274,6 +3431,7 @@ fn run(options: &CliOptions, once: bool, print_only: bool) -> Result<()> { .max() .unwrap_or(0); let mut next_update_check = Instant::now() + update_check_jitter(&config.token); + let mut update_checkpoint = None; let mut next_stats_log = Instant::now() + RUNTIME_STATS_INTERVAL; let clock = Arc::new(Mutex::new(ClockCalibration::default())); let live = (!print_only) @@ -3306,9 +3464,6 @@ fn run(options: &CliOptions, once: bool, print_only: bool) -> Result<()> { if apply_remote(&mut config, &remote) { next_latency.clear(); } - if config.collect_interval != previous_collect_interval { - next_collect = Instant::now() + Duration::from_secs(config.collect_interval); - } if config.collect_interval != previous_collect_interval || config.report_interval != previous_report_interval { @@ -3322,14 +3477,23 @@ fn run(options: &CliOptions, once: bool, print_only: bool) -> Result<()> { return Err("WSS report persistence acknowledgement timed out".into()); } - // Check after ACK pruning, before the next sample refills the pending queue. + // Continuous sampling need not leave the queue empty. Wait for the samples + // present when the check became due, and preserve newer ones across restart. + if config.auto_update && Instant::now() >= next_update_check { + update_checkpoint.get_or_insert(last_emitted_timestamp); + } if !once && !print_only && config.auto_update - && pending_samples.is_empty() - && Instant::now() >= next_update_check + && update_checkpoint.is_some_and(|timestamp| { + live.as_ref() + .is_some_and(|sender| sender.persisted_through() >= timestamp) + }) { - match update(&agent, &config.agent_mirror) { + update_checkpoint = None; + match rewrite_pending_spool(&spool_path, &pending_samples) + .and_then(|()| update(&agent, &config.agent_mirror)) + { Ok(true) => return Ok(()), Ok(false) => next_update_check = Instant::now() + UPDATE_CHECK_INTERVAL, Err(error) => { @@ -3415,11 +3579,7 @@ fn run(options: &CliOptions, once: bool, print_only: bool) -> Result<()> { { eprintln!("pending report spool compaction failed: {error}"); } - next_collect = advance_deadline( - next_collect, - Duration::from_secs(config.collect_interval), - Instant::now(), - ); + next_collect = advance_deadline(next_collect, SAMPLE_INTERVAL, Instant::now()); } if Instant::now() >= next_stats_log { @@ -3432,7 +3592,7 @@ fn run(options: &CliOptions, once: bool, print_only: bool) -> Result<()> { } else { next_collect }; - let maintenance_wake_at = if config.auto_update && pending_samples.is_empty() { + let maintenance_wake_at = if config.auto_update && update_checkpoint.is_none() { next_stats_log.min(next_update_check) } else { next_stats_log @@ -3469,18 +3629,19 @@ mod tests { use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use super::{ - CLOCK_CALIBRATION_MAX_AGE, CapturedOutput, CliOptions, ClockCalibration, + CLOCK_CALIBRATION_MAX_AGE, CapturedOutput, CliOptions, ClockCalibration, DiskMetric, GithubReleaseAsset, LatencyResult, LatencyTask, LiveAck, MAX_PENDING_LATENCY_RESULTS, 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, - advance_deadline, clock_offset_from_http_date, corrected_timestamp, + advance_deadline, clock_offset_from_http_date, corrected_timestamp, dedupe_disks, 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, parse_system_profiler_gpu_names, ping_latencies, ping_latency, prune_report_samples, - release_asset_sha256, remote_result_text, sanitize_latency_tasks, update_check_jitter, - valid_endpoint, version_triplet, write_remote_task_journal, + release_asset_sha256, remote_result_text, sanitize_latency_tasks, selected_interface, + update_check_jitter, valid_endpoint, version_triplet, wildcard_match, + write_remote_task_journal, }; #[cfg(any(target_os = "windows", target_os = "macos", target_os = "freebsd"))] @@ -3565,6 +3726,80 @@ mod tests { assert!(!disk_device("nvme0n1p1")); } + #[test] + fn filters_virtual_interfaces_and_supports_include_exclude_patterns() { + assert!(!selected_interface("lo", "")); + assert!(!selected_interface("docker0", "")); + assert!(!selected_interface("veth123", "")); + assert!(!selected_interface("br0", "!eth*")); + assert!(selected_interface("br0", "br0")); + assert!(!selected_interface("br0", "br*,!br0")); + assert!(!selected_interface("br0", "!br0,br*")); + assert!(!selected_interface("Software Loopback Interface 1", "")); + assert!(selected_interface("eth0", "")); + assert!(selected_interface("ens3", "eth*,ens*")); + assert!(!selected_interface("wlan0", "eth*,ens*")); + assert!(!selected_interface("eth0", "eth*,!eth0")); + assert!(selected_interface("ens3", "!eth*")); + assert!(wildcard_match("en?3", "ens3")); + assert!(!wildcard_match("en?3", "enp4s0")); + } + + #[test] + fn deduplicates_mounts_for_the_same_device() { + let disks = dedupe_disks([ + DiskMetric { + name: "/dev/vg/root".into(), + mount_point: "/var".into(), + total: 10, + used: 4, + ..DiskMetric::default() + }, + DiskMetric { + name: "/dev/vg/root".into(), + mount_point: "/".into(), + total: 20, + used: 8, + ..DiskMetric::default() + }, + DiskMetric { + name: "tank/data".into(), + mount_point: "/data".into(), + total: 30, + used: 9, + ..DiskMetric::default() + }, + DiskMetric { + name: "tank/archive".into(), + mount_point: "/archive".into(), + total: 25, + used: 7, + ..DiskMetric::default() + }, + DiskMetric { + name: "/var/lib/data".into(), + mount_point: "/var/lib/data".into(), + total: 40, + used: 12, + ..DiskMetric::default() + }, + ]); + assert_eq!(disks.len(), 3); + assert_eq!(disks[0].mount_point, "/"); + assert_eq!(disks[0].total, 20); + assert_eq!(disks[1].name, "tank/data"); + assert_eq!(disks[2].name, "/var/lib/data"); + } + + #[cfg(target_os = "linux")] + #[test] + fn keeps_root_filesystem_but_filters_virtual_mounts() { + assert!(!super::excluded_filesystem("overlay", "/")); + assert!(super::excluded_filesystem("tmpfs", "/tmp")); + assert!(super::excluded_filesystem("proc", "/proc")); + assert!(!super::excluded_filesystem("btrfs", "/var")); + } + #[test] fn parses_release_versions() { assert_eq!(normalized_version("v1.2.3"), "1.2.3"); diff --git a/backend/src/db/queries.rs b/backend/src/db/queries.rs index 73d2a08..a81e500 100644 --- a/backend/src/db/queries.rs +++ b/backend/src/db/queries.rs @@ -346,7 +346,7 @@ pub async fn list_servers_with_live( report_interval: row.try_get("report_interval")?, collect_interval: row .try_get::("collect_interval")? - .max(nodeflare_telemetry::MIN_COLLECT_INTERVAL as i64), + .max(nodeflare_telemetry::MIN_UPLOAD_INTERVAL as i64), rx_correction: row.try_get("rx_correction")?, tx_correction: row.try_get("tx_correction")?, agent_mirror: row.try_get("agent_mirror")?, @@ -599,7 +599,7 @@ pub async fn agent_config(db: &Database, id: &str) -> Result("report_interval")?, - "collect_interval": row.try_get::("collect_interval")?.max(nodeflare_telemetry::MIN_COLLECT_INTERVAL as i64), + "collect_interval": row.try_get::("collect_interval")?.max(nodeflare_telemetry::MIN_UPLOAD_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, @@ -2817,6 +2817,36 @@ mod tests { assert_eq!(latest.latency[0].latency_ms, -1.0); } + #[tokio::test] + async fn live_buffer_preserves_each_one_second_sample_for_dashboard_playback() { + use crate::websocket::ingest::AgentBuffer; + let (db, identity) = test_server().await; + let start = now() - 10; + let mut buffer = AgentBuffer::new(&db, &identity.server_id).await.unwrap(); + let reports = (0..3) + .map(|index| sample(start + index, 10.0 + index as f64)) + .collect::>(); + let result = buffer + .receive(&db, &identity, "", reports.clone(), false) + .await + .unwrap(); + assert_eq!( + result + .samples + .iter() + .map(|report| (report.timestamp, report.cpu)) + .collect::>(), + vec![(start, 10.0), (start + 1, 11.0), (start + 2, 12.0)] + ); + assert_eq!(result.latest.unwrap().timestamp, start + 2); + assert!(result.acknowledgement.is_none()); + let duplicate = buffer + .receive(&db, &identity, "", reports, false) + .await + .unwrap(); + assert!(duplicate.samples.is_empty()); + } + #[tokio::test] async fn live_buffer_commits_without_reupload_and_replays_only_unconfirmed_samples() { use crate::websocket::ingest::AgentBuffer; diff --git a/backend/src/routes/admin.rs b/backend/src/routes/admin.rs index f7c7cab..6b773a6 100644 --- a/backend/src/routes/admin.rs +++ b/backend/src/routes/admin.rs @@ -872,8 +872,7 @@ fn validate_server(input: &ServerInput) -> Result<(), &'static str> { } if !(1..=31).contains(&input.reset_day) || !(15..=3600).contains(&input.report_interval) - || !(nodeflare_telemetry::MIN_COLLECT_INTERVAL as i64..=60) - .contains(&input.collect_interval) + || !(nodeflare_telemetry::MIN_UPLOAD_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 { diff --git a/backend/src/websocket/agent.rs b/backend/src/websocket/agent.rs index 1a2781c..6c31a17 100644 --- a/backend/src/websocket/agent.rs +++ b/backend/src/websocket/agent.rs @@ -211,7 +211,7 @@ async fn run( .get(&identity.server_id) .is_none_or(|current| current.timestamp < report.timestamp) { - broadcast_report(&state, &identity, &report); + broadcast_reports(&state, &identity, &result.samples); let mut cached = report; if let Some(previous) = live.get(&identity.server_id) { cached @@ -398,31 +398,37 @@ fn send_persistence_error(outbound: &mpsc::Sender, persisted: i64) )); } -fn broadcast_report(state: &AppState, identity: &AgentIdentity, report: &AgentReport) { +fn broadcast_reports(state: &AppState, identity: &AgentIdentity, reports: &[AgentReport]) { if identity.hidden || state.dashboard_tx.receiver_count() == 0 { return; } - 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 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 = telemetry::encode(&serde_json::json!({ "type": "batchUpdate", "ts": crate::db::now(), - "updates": [{"serverId": identity.server_id, "samples": [{"ts": report.timestamp, "data": data}]}], + "updates": [{"serverId": identity.server_id, "samples": samples}], })); if let Ok(payload) = payload { let _ = state.dashboard_tx.send(DashboardEvent { diff --git a/backend/src/websocket/ingest.rs b/backend/src/websocket/ingest.rs index 12e1b5d..42080d6 100644 --- a/backend/src/websocket/ingest.rs +++ b/backend/src/websocket/ingest.rs @@ -16,6 +16,7 @@ pub struct AgentBuffer { pub struct IngestResult { pub latest: Option, + pub samples: Vec, pub acknowledgement: Option, } @@ -62,15 +63,16 @@ impl AgentBuffer { { acknowledgement = Some(self.flush(db, identity, remote_ip).await?); } - let mut latest = None; + let mut samples = Vec::with_capacity(reports.len()); 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); + latency.extend(live.latency_results.iter().cloned()); + samples.push(live); } + let mut latest = samples.last().cloned(); if let Some(latest) = &mut latest { latest.latency_results = latency; } @@ -81,6 +83,7 @@ impl AgentBuffer { } Ok(IngestResult { latest, + samples, acknowledgement, }) } diff --git a/frontend/src/AppProvider.tsx b/frontend/src/AppProvider.tsx index e2feec1..e493f88 100644 --- a/frontend/src/AppProvider.tsx +++ b/frontend/src/AppProvider.tsx @@ -3,9 +3,9 @@ import { api, ApiError } from "./api"; import { demoConfig, demoExchangeRates, demoServers } from "./demo"; import { applyBatch, + createLivePlayback, mergeServerLive, pruneLiveMetrics, - type BatchUpdate, type LiveMetricsMap, } from "./live"; import { ui } from "./locale"; @@ -74,7 +74,7 @@ export function AppProvider({ children }: { children: ReactNode }) { const [servers, setServers] = useState([]); const [liveMetrics, setLiveMetrics] = useState({}); const [liveConnected, setLiveConnected] = useState(false); - const pendingLive = useRef([]); + const livePlayback = useRef(createLivePlayback()); const [clockNow, setClockNow] = useState(() => Date.now()); const [exchangeRates, setExchangeRates] = useState(null); const [loading, setLoading] = useState(true); @@ -222,15 +222,17 @@ export function AppProvider({ children }: { children: ReactNode }) { let timer: number | undefined; const tick = () => { setClockNow(Date.now()); - if (!pendingLive.current.length) return; - const batch = pendingLive.current; - pendingLive.current = []; + const batch = livePlayback.current.take(serversRef.current); + if (!batch.length) return; setLiveMetrics((current) => applyBatch(current, batch, serversRef.current)); }; const sync = () => { window.clearInterval(timer); timer = undefined; - if (document.hidden || navigator.onLine === false) return; + if (document.hidden || navigator.onLine === false) { + livePlayback.current.clear(); + return; + } tick(); timer = window.setInterval(tick, LIVE_CARD_REFRESH_INTERVAL_MS); }; @@ -251,13 +253,13 @@ export function AppProvider({ children }: { children: ReactNode }) { let active = true; const disconnect = connectLive({ serverId: selectedId }, { onBatch: (updates) => { - pendingLive.current.push(...updates); + livePlayback.current.enqueue(updates, serversRef.current); }, onConnectedChange: (connected) => { liveConnectedRef.current = connected; setLiveConnected(connected); if (!connected) { - pendingLive.current = []; + livePlayback.current.clear(); } if (active && !connected && !document.hidden && navigator.onLine !== false) { void reload(true); @@ -266,7 +268,7 @@ export function AppProvider({ children }: { children: ReactNode }) { }); return () => { active = false; - pendingLive.current = []; + livePlayback.current.clear(); disconnect(); }; }, [access, configReady, reload, selectedId]); diff --git a/frontend/src/components/AdminPanel.tsx b/frontend/src/components/AdminPanel.tsx index 2bbaa86..2b15378 100644 --- a/frontend/src/components/AdminPanel.tsx +++ b/frontend/src/components/AdminPanel.tsx @@ -1309,8 +1309,8 @@ export function AdminPanel({
-
-
+
+
updateForm("auto_renewal", value)} /> updateForm("auto_update", value)} /> updateForm("hidden", value)} /> updateForm("offline_notify_disabled", value)} />
: null} diff --git a/frontend/src/live.test.ts b/frontend/src/live.test.ts index 6a802b5..e933f43 100644 --- a/frontend/src/live.test.ts +++ b/frontend/src/live.test.ts @@ -1,6 +1,7 @@ import { describe, expect, test } from "bun:test"; import { applyBatch, + createLivePlayback, liveLatencySamples, mergeLiveLatency, mergeLiveResults, @@ -130,6 +131,90 @@ describe("applyBatch", () => { }); }); +describe("one-second card playback", () => { + test("renders every real CPU and network sample from successive three-second uploads", () => { + const queue = createLivePlayback(); + const servers = [server()]; + let current: LiveMetricsMap = {}; + const displayed = []; + for (let tick = 0; tick < 9; tick++) { + const now = (1_003 + tick) * 1_000; + if (tick % 3 === 0) { + queue.enqueue([{ + serverId: "s1", + samples: [1, 2, 3].map((offset) => ({ + ts: 1_000 + tick + offset, + data: { cpu: tick + offset, net_in: (tick + offset) * 100, net_out: (tick + offset) * 200 }, + })), + }], servers, now); + } + current = applyBatch(current, queue.take(servers, now), servers); + const card = mergeServerLive(servers[0], current.s1, now, 180); + displayed.push([card.cpu, card.net_in, card.net_out]); + } + expect(displayed).toEqual(Array.from({ length: 9 }, (_, index) => [ + index + 1, (index + 1) * 100, (index + 1) * 200, + ])); + expect(queue.take(servers, 1_012_000)).toEqual([]); + }); + + test("deduplicates overlapping batches and never plays samples backwards", () => { + const queue = createLivePlayback(); + const servers = [server()]; + const updates = [{ serverId: "s1", samples: [1_003, 1_001, 1_002, 1_002].map((ts) => ({ ts, data: { cpu: ts } })) }]; + queue.enqueue(updates, servers, 0); + expect(queue.take(servers, 0)[0].samples?.[0].ts).toBe(1_001); + queue.enqueue(updates, servers, 1_000); + expect(queue.take(servers, 1_000)[0].samples?.[0].ts).toBe(1_002); + expect(queue.take(servers, 2_000)[0].samples?.[0].ts).toBe(1_003); + queue.enqueue(updates, servers, 3_000); + expect(queue.take(servers, 3_000)).toEqual([]); + }); + + test("bounds replay delay after an outage while retaining latency results", () => { + const queue = createLivePlayback(); + const servers = [server()]; + queue.enqueue([{ + serverId: "s1", + samples: Array.from({ length: 100 }, (_, index) => ({ + ts: 1_001 + index, + data: { cpu: index, latency_results: [{ task_id: "t1", timestamp: 1_001 + index, latency_ms: index, packet_loss: 0 }] }, + })), + }], servers, 0); + const next = queue.take(servers, 0)[0].samples![0]; + expect(next.ts).toBe(1_098); + expect(next.data.latency_results).toHaveLength(100); + expect(queue.take(servers, 1_000)[0].samples?.[0].ts).toBe(1_099); + expect(queue.take(servers, 2_000)[0].samples?.[0].ts).toBe(1_100); + expect(queue.take(servers, 3_000)).toEqual([]); + }); + + test("skips samples superseded by bootstrap without dropping latency", () => { + const queue = createLivePlayback(); + queue.enqueue([{ serverId: "s1", samples: [{ + ts: 1_001, data: { cpu: 99, latency_results: [{ task_id: "t1", timestamp: 1_001, latency_ms: 20, packet_loss: 0 }] }, + }] }], [server()], 0); + const servers = [server({ timestamp: 1_005, cpu: 10 })]; + const state = applyBatch({}, queue.take(servers, 0), servers); + expect(mergeServerLive(servers[0], state.s1, 0, 180).cpu).toBe(10); + expect(state.s1.latencyResults).toHaveLength(1); + }); + + test("clears suspended queues, removes deleted nodes and drops stale metrics", () => { + const queue = createLivePlayback(); + const servers = [server()]; + const updates = [{ serverId: "s1", samples: [{ ts: 1_001, data: { cpu: 20 } }] }]; + queue.enqueue(updates, servers, 0); + queue.clear(); + expect(queue.take(servers, 0)).toEqual([]); + queue.enqueue(updates, servers, 0); + expect(queue.take([], 0)).toEqual([]); + expect(queue.take(servers, 0)).toEqual([]); + queue.enqueue(updates, servers, 0); + expect(queue.take(servers, 6_000)).toEqual([]); + }); +}); + describe("mergeServerLive", () => { test("applies a newer live sample and ticks uptime", () => { const merged = mergeServerLive( diff --git a/frontend/src/live.ts b/frontend/src/live.ts index 100634d..929df7b 100644 --- a/frontend/src/live.ts +++ b/frontend/src/live.ts @@ -19,6 +19,74 @@ export interface BatchUpdate { } const MAX_LIVE_LATENCY_RESULTS = 4096; +const MAX_PLAYBACK_SAMPLES = 3; + +// Keep a short queue of real samples so a three-second upload can drive +// one-second card updates. Long reconnect backlogs must not delay the live view. +export function createLivePlayback() { + const queues = new Map(); + return { + enqueue(updates: readonly BatchUpdate[], servers: readonly Server[], now = Date.now()) { + const known = new Map(servers.map((server) => [server.id, server])); + for (const update of updates) { + if (!update?.serverId || !Array.isArray(update.samples)) continue; + const server = known.get(update.serverId); + if (!server) continue; + const incoming = validSamples(update.samples); + if (!incoming.length) continue; + const queue = queues.get(server.id) ?? { + samples: [], playedThrough: 0, latencyResults: [], receivedAt: now, + }; + const floor = Math.max(queue.playedThrough, server.timestamp ?? 0); + const samples = new Map(queue.samples.map((sample) => [sample.ts, sample])); + for (const sample of incoming) { + if (sample.ts > floor) samples.set(sample.ts, sample); + } + queue.samples = [...samples.values()] + .filter((sample) => sample.ts > floor) + .sort((left, right) => left.ts - right.ts) + .slice(-MAX_PLAYBACK_SAMPLES); + queue.latencyResults = mergeLiveResults(queue.latencyResults, + incoming.flatMap((sample) => Array.isArray(sample.data.latency_results) + ? sample.data.latency_results : [])); + queue.receivedAt = now; + queues.set(server.id, queue); + } + }, + take(servers: readonly Server[], now = Date.now()): BatchUpdate[] { + const known = new Map(servers.map((server) => [server.id, server])); + const updates: BatchUpdate[] = []; + for (const [serverId, queue] of queues) { + const server = known.get(serverId); + if (!server) { + queues.delete(serverId); + continue; + } + const floor = Math.max(queue.playedThrough, server.timestamp ?? 0); + queue.samples = now - queue.receivedAt > 5_000 + ? [] : queue.samples.filter((sample) => sample.ts > floor); + const sample = queue.samples.shift(); + if (!sample && !queue.latencyResults.length) continue; + if (sample) queue.playedThrough = sample.ts; + updates.push({ + serverId, + samples: [{ + ts: sample?.ts ?? Math.max(floor, 1), + data: { ...sample?.data, latency_results: queue.latencyResults }, + }], + }); + queue.latencyResults = []; + } + return updates; + }, + clear() { queues.clear(); }, + }; +} export function mergeLiveResults( previous: LiveLatencyResult[] | undefined, @@ -79,6 +147,11 @@ function sampleMetrics(sample: LiveSample) { return metrics; } +function validSamples(samples: readonly LiveSample[]) { + return samples.filter((sample) => sample && Number.isFinite(sample.ts) && sample.ts > 0 + && sample.data && typeof sample.data === "object" && !Array.isArray(sample.data)); +} + export function applyBatch( current: LiveMetricsMap, updates: readonly BatchUpdate[], @@ -90,9 +163,7 @@ export function applyBatch( if (!update?.serverId || !Array.isArray(update.samples) || !update.samples.length) continue; const server = knownServers.get(update.serverId); if (!server) continue; - const samples = update.samples - .filter((sample) => sample && Number.isFinite(sample.ts) && sample.ts > 0 - && sample.data && typeof sample.data === "object" && !Array.isArray(sample.data)) + const samples = validSamples(update.samples) .sort((left, right) => left.ts - right.ts); if (!samples.length) continue; const previous = next[update.serverId]; diff --git a/scripts/agent-runtime-test.mjs b/scripts/agent-runtime-test.mjs index b7fe1c6..f1d69d5 100644 --- a/scripts/agent-runtime-test.mjs +++ b/scripts/agent-runtime-test.mjs @@ -73,7 +73,7 @@ try { proxy.listen(0, "127.0.0.1"); await once(proxy, "listening"); - console.log("agent runtime: automatic update during three-second sampling"); + console.log("agent runtime: automatic update during one-second sampling"); websocketServer = new WebSocketServer({ noServer: true }); websocketServer.on("headers", (headers) => { headers.push("X-NodeFlare-Agent-Protocol: 1"); @@ -131,7 +131,7 @@ try { } assert(token, "Unable to create deterministic update jitter"); const updating = startAgent(upgradeServer.address().port, token, "update"); - await waitUntil(() => releaseRequests.length > 0, 12_000, "Automatic update was starved by sampling"); + await waitUntil(() => releaseRequests.length > 0, 22_000, "Automatic update was starved by sampling"); assert(samples > 0, "Update check must follow acknowledged samples"); assert(redirected, "Same-origin WebSocket redirects must remain supported"); await stopAgent(updating); @@ -139,11 +139,12 @@ try { 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 }; + const connection = { timestamps: [], frames: [], 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()); + if (message.samples.length) connection.frames.push({ at: Date.now(), count: message.samples.length }); for (const sample of message.samples) { if (!connection.timestamps.length) connection.firstInfo = Boolean(sample.info); connection.infoCount += Number(Boolean(sample.info)); @@ -154,7 +155,7 @@ try { connection.durable = connection.timestamps.at(-1); connection.commits++; } - if (telemetryConnections.length === 1 && connection.timestamps.length === 3) socket.terminate(); + if (telemetryConnections.length === 1 && connection.frames.length >= 3) socket.terminate(); }); }; websocketServer.on("connection", trackTelemetry); @@ -168,7 +169,11 @@ try { 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"); + assert(before.timestamps.slice(1).every((timestamp, index) => timestamp - before.timestamps[index] === 1), + "CPU/network samples must be one second apart"); + assert(before.frames.slice(1).every((frame, index) => frame.at - before.frames[index].at >= 2_800), + "Sampling every second must not cause per-second uploads"); + assert(before.frames.some((frame) => frame.count >= 3), "Uploads must batch the one-second samples"); console.log("agent runtime: remote command survives disconnect and returns its result"); const taskId = randomUUID(); diff --git a/scripts/websocket-smoke.mjs b/scripts/websocket-smoke.mjs index 7ce629b..a0b24b7 100644 --- a/scripts/websocket-smoke.mjs +++ b/scripts/websocket-smoke.mjs @@ -710,8 +710,10 @@ try { ) { 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"); + assert.deepEqual(update.updates[0].samples.map((sample) => sample.ts), + samples.map((sample) => sample.timestamp), "Browser receives every real sample in order"); + assert.equal(update.updates[0].samples.flatMap((sample) => sample.data.latency_results).length, + 2, "All new latency results are retained"); const publicIdentityFields = [ "cpu_model", diff --git a/shared/telemetry.rs b/shared/telemetry.rs index f45d8f3..6b8475e 100644 --- a/shared/telemetry.rs +++ b/shared/telemetry.rs @@ -2,7 +2,7 @@ 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 MIN_UPLOAD_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;