diff --git a/agent/Cargo.lock b/agent/Cargo.lock index 4f7280e..0638df5 100644 --- a/agent/Cargo.lock +++ b/agent/Cargo.lock @@ -476,6 +476,7 @@ version = "1.0.0" dependencies = [ "clap", "httpdate", + "libc", "nodeflare-telemetry", "serde", "serde_json", diff --git a/agent/Cargo.toml b/agent/Cargo.toml index 455e270..7477ae0 100644 --- a/agent/Cargo.toml +++ b/agent/Cargo.toml @@ -17,6 +17,9 @@ ureq = { version = "3.4", default-features = false, features = ["json", "rustls" url = "2.5" tungstenite = { version = "0.30", default-features = false, features = ["handshake", "rustls-tls-webpki-roots"] } +[target.'cfg(target_os = "linux")'.dependencies] +libc = "0.2" + [target.'cfg(any(target_os = "windows", target_os = "macos", target_os = "freebsd"))'.dependencies] sysinfo = { version = "0.39.6", default-features = false, features = ["disk", "network", "system"] } diff --git a/agent/src/main.rs b/agent/src/main.rs index 59fd659..2e38334 100644 --- a/agent/src/main.rs +++ b/agent/src/main.rs @@ -1,7 +1,10 @@ mod live_batch; mod runtime_stats; use nodeflare_telemetry as telemetry; -use telemetry::{DiskMetric, GpuMetric, LatencyResult, Report}; +use telemetry::{ + AGENT_CAPABILITIES_HEADER, AGENT_PROTOCOL_HEADER, AGENT_PROTOCOL_VERSION, DiskMetric, + GpuMetric, LatencyResult, REQUIRED_AGENT_CAPABILITIES, Report, +}; use std::collections::{HashMap, HashSet, VecDeque}; use std::env; @@ -78,8 +81,6 @@ const PUBLIC_IP_V6_URL: &str = "https://ipv6.icanhazip.com/"; const RUNTIME_STATS_INTERVAL: Duration = Duration::from_secs(5 * 60); const CLOCK_CALIBRATION_MAX_AGE: Duration = Duration::from_secs(24 * 60 * 60); const CLOCK_CALIBRATION_MIN_CHANGE_MS: i64 = 20_000; -const AGENT_PROTOCOL_VERSION: &str = "1"; -const AGENT_CAPABILITIES: &str = "metrics-v1,config-v1,remote-exec-v1,task-ack-v1"; type Error = Box; type Result = std::result::Result; @@ -196,6 +197,30 @@ impl LatencyExecutor { } results } + + fn schedule( + &mut self, + tasks: &[LatencyTask], + deadlines: &mut HashMap, + now: Instant, + ) { + for task in tasks { + if deadlines + .get(&task.id) + .is_some_and(|deadline| *deadline > now) + { + continue; + } + let delay = if self.enqueue(task.clone()) { + Duration::from_secs(task.interval_seconds.clamp(30, 3600)) + } else { + // A queued/running task can outlive its interval. Never leave an + // expired deadline behind for the main loop to spin on. + SAMPLE_INTERVAL + }; + deadlines.insert(task.id.clone(), now + delay); + } + } } #[derive(Debug, Deserialize)] @@ -695,39 +720,90 @@ fn connection_counts() -> (i64, i64) { #[cfg(target_os = "linux")] fn disk_usage() -> Vec { - let output = command( - "df", - &["-B1", "-l", "--output=source,target,fstype,size,used"], - ); - output - .lines() - .skip(1) - .filter_map(|line| { - let fields = line.split_whitespace().collect::>(); - if fields.len() < 5 { + disk_mounts(&text("/proc/self/mounts")) + .filter_map(|(name, mount_point)| { + // Ignore file bind mounts such as a container's /etc/hosts. + if !Path::new(&mount_point).is_dir() { return None; } - 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: source.to_string(), + let (total, used) = filesystem_space(&mount_point)?; + Some(DiskMetric { + name, mount_point, used, total, ..DiskMetric::default() }) }) - .collect::>() + .collect() +} + +#[cfg(target_os = "linux")] +fn disk_mounts(contents: &str) -> impl Iterator + '_ { + fn unescape(value: &str) -> String { + value + .replace("\\040", " ") + .replace("\\011", "\t") + .replace("\\012", "\n") + .replace("\\134", "\\") + } + contents.lines().filter_map(|line| { + let mut fields = line.split_whitespace(); + let source = unescape(fields.next()?); + let mount_point = unescape(fields.next()?); + let filesystem = fields.next()?; + if (!source.starts_with('/') && source.contains(':')) + || source.starts_with("//") + || excluded_filesystem(filesystem, &mount_point) + { + return None; + } + Some((source, mount_point)) + }) +} + +#[cfg(target_os = "linux")] +fn filesystem_space(mount_point: &str) -> Option<(i64, i64)> { + let path = std::ffi::CString::new(mount_point).ok()?; + let mut stat = std::mem::MaybeUninit::::uninit(); + // SAFETY: path is NUL-terminated and stat points to writable storage of the + // platform's statvfs type, including both glibc and musl layouts. + if unsafe { libc::statvfs(path.as_ptr(), stat.as_mut_ptr()) } != 0 { + return None; + } + // SAFETY: statvfs returned success and initialized stat. + let stat = unsafe { stat.assume_init() }; + let block_size = u128::from(if stat.f_frsize == 0 { + stat.f_bsize + } else { + stat.f_frsize + }); + let total = (u128::from(stat.f_blocks) * block_size).min(i64::MAX as u128) as i64; + let used = (u128::from(stat.f_blocks.saturating_sub(stat.f_bfree)) * block_size) + .min(i64::MAX as u128) as i64; + (total > 0).then_some((total, used)) } #[cfg(target_os = "linux")] fn excluded_filesystem(filesystem: &str, mount_point: &str) -> bool { + // Remote/autofs mounts can block indefinitely while their server is down. + if matches!( + filesystem, + "nfs" + | "nfs4" + | "cifs" + | "smb3" + | "autofs" + | "9p" + | "ceph" + | "afs" + | "fuse.sshfs" + | "fuse.glusterfs" + | "fuse.rclone" + | "fuse.s3fs" + ) { + return true; + } let mount_point = mount_point.to_ascii_lowercase(); // Keep the root filesystem visible even when a container reports it as overlay. if mount_point == "/" { @@ -1831,13 +1907,12 @@ fn connect_live(endpoint: &str, token: &str) -> Result<(LiveSocket, Option) request .headers_mut() .insert("User-Agent", format!("nodeflare-agent/{VERSION}").parse()?); + request + .headers_mut() + .insert(AGENT_PROTOCOL_HEADER, AGENT_PROTOCOL_VERSION.parse()?); request.headers_mut().insert( - "X-NodeFlare-Agent-Protocol", - AGENT_PROTOCOL_VERSION.parse()?, - ); - request.headers_mut().insert( - "X-NodeFlare-Agent-Capabilities", - AGENT_CAPABILITIES.parse()?, + AGENT_CAPABILITIES_HEADER, + REQUIRED_AGENT_CAPABILITIES.join(",").parse()?, ); let stream = connect_live_stream(&url, deadline)?; match client_tls(request, stream) { @@ -1869,7 +1944,7 @@ fn connect_live(endpoint: &str, token: &str) -> Result<(LiveSocket, Option) .and_then(|value| clock_offset_from_http_date(value, started_ms, ended_ms)); if !response .headers() - .get("x-nodeflare-agent-protocol") + .get(AGENT_PROTOCOL_HEADER) .is_some_and(|value| value == AGENT_PROTOCOL_VERSION) { return Err(format!( @@ -3503,27 +3578,7 @@ fn run(options: &CliOptions, once: bool, print_only: bool) -> Result<()> { } } pending_results.extend(latency_executor.drain()); - let current = Instant::now(); - let due = config - .latency_tasks - .iter() - .filter(|task| { - next_latency - .get(&task.id) - .is_none_or(|deadline| *deadline <= current) - }) - .cloned() - .collect::>(); - if !due.is_empty() { - let scheduled_at = Instant::now(); - for task in due { - let task_id = task.id.clone(); - let interval = task.interval_seconds.clamp(30, 3600); - if latency_executor.enqueue(task) { - next_latency.insert(task_id, scheduled_at + Duration::from_secs(interval)); - } - } - } + latency_executor.schedule(&config.latency_tasks, &mut next_latency, Instant::now()); if Instant::now() >= next_collect && (!once || once_target_timestamp.is_none()) { let offset_ms = shared_clock_offset(&clock); @@ -3532,15 +3587,17 @@ fn run(options: &CliOptions, once: bool, print_only: bool) -> Result<()> { result.timestamp = corrected_timestamp(result.timestamp, offset_ms); latest_results.insert(result.task_id.clone(), result); } - let collection_started = Instant::now(); - let mut report = collector.collect(&config, latest_results.into_values().collect(), 0); - stats.collection_finished(collection_started.elapsed()); - let persisted_through = live.as_ref().map_or(0, LiveSender::persisted_through); - report.timestamp = monotonic_report_timestamp( + // CPU and network counters are read at the beginning of collect. + // A slow GPU/disk refresh must not shift their timestamp to a later second. + let timestamp = monotonic_report_timestamp( corrected_timestamp(unix_timestamp(), offset_ms), last_emitted_timestamp, - persisted_through, + live.as_ref().map_or(0, LiveSender::persisted_through), ); + let collection_started = Instant::now(); + let report = + collector.collect(&config, latest_results.into_values().collect(), timestamp); + stats.collection_finished(collection_started.elapsed()); last_emitted_timestamp = report.timestamp; if print_only { println!("{}", serde_json::to_string_pretty(&report)?); @@ -3800,6 +3857,98 @@ mod tests { assert!(!super::excluded_filesystem("btrfs", "/var")); } + #[cfg(target_os = "linux")] + #[test] + fn reads_local_mounts_and_decodes_escaped_paths_once() { + let mounts = super::disk_mounts( + r"overlay / overlay rw 0 0 +proc /proc proc rw 0 0 +tmpfs /run tmpfs rw 0 0 +/dev/vda1 /data\040volume ext4 rw 0 0 +/dev/vda2 /literal\134040path ext4 rw 0 0 +/dev/vda3 /tab\011dir ext4 rw 0 0 +server:/export /remote nfs4 rw 0 0 +//server/share /remote-share cifs rw 0 0 +none /auto autofs rw 0 0 +invalid", + ) + .collect::>(); + assert_eq!( + mounts, + vec![ + ("overlay".into(), "/".into()), + ("/dev/vda1".into(), "/data volume".into()), + ("/dev/vda2".into(), "/literal\\040path".into()), + ("/dev/vda3".into(), "/tab\tdir".into()), + ] + ); + } + + #[cfg(target_os = "linux")] + #[test] + fn reads_root_disk_capacity_without_external_commands() { + let (total, used) = super::filesystem_space("/").unwrap(); + assert!(total > 0); + assert!((0..=total).contains(&used)); + assert!( + super::disk_usage() + .iter() + .any(|disk| disk.mount_point == "/") + ); + assert!(super::filesystem_space("/proc/self/not-a-mount").is_none()); + assert!(super::filesystem_space("/nul\0path").is_none()); + } + + #[test] + fn overdue_latency_tasks_wait_and_resume_without_duplicate_probes() { + let (task_tx, task_rx) = std::sync::mpsc::channel(); + let (result_tx, result_rx) = std::sync::mpsc::channel(); + let mut executor = super::LatencyExecutor { + task_tx, + result_rx, + in_flight: Default::default(), + stats: Default::default(), + }; + let task = LatencyTask { + id: "queued-probe".into(), + name: "probe".into(), + task_type: "tcp".into(), + target: "example.com".into(), + port: Some(443), + interval_seconds: 30, + }; + let mut deadlines = std::collections::HashMap::new(); + let started = Instant::now(); + executor.schedule(std::slice::from_ref(&task), &mut deadlines, started); + assert_eq!(task_rx.try_recv().unwrap().id, task.id); + assert_eq!(deadlines[&task.id], started + Duration::from_secs(30)); + + let overdue = started + Duration::from_secs(30); + executor.schedule(std::slice::from_ref(&task), &mut deadlines, overdue); + assert!(task_rx.try_recv().is_err()); + assert_eq!(deadlines[&task.id], overdue + Duration::from_secs(1)); + + result_tx + .send(LatencyResult { + task_id: task.id.clone(), + ..LatencyResult::default() + }) + .unwrap(); + assert_eq!(executor.drain().len(), 1); + let retry_at = deadlines[&task.id]; + executor.schedule(std::slice::from_ref(&task), &mut deadlines, retry_at); + executor.schedule(std::slice::from_ref(&task), &mut deadlines, retry_at); + assert_eq!(task_rx.try_recv().unwrap().id, task.id); + assert!(task_rx.try_recv().is_err()); + assert_eq!(deadlines[&task.id], retry_at + Duration::from_secs(30)); + + drop(task_rx); + executor.in_flight.clear(); + let retry_at = deadlines[&task.id]; + executor.schedule(std::slice::from_ref(&task), &mut deadlines, retry_at); + assert_eq!(deadlines[&task.id], retry_at + Duration::from_secs(1)); + } + #[test] fn parses_release_versions() { assert_eq!(normalized_version("v1.2.3"), "1.2.3"); diff --git a/backend/src/websocket/agent.rs b/backend/src/websocket/agent.rs index 6c31a17..4b3dc1d 100644 --- a/backend/src/websocket/agent.rs +++ b/backend/src/websocket/agent.rs @@ -12,15 +12,14 @@ use nodeflare_telemetry as telemetry; use serde::Deserialize; use std::net::SocketAddr; use std::sync::Arc; +use telemetry::{ + AGENT_CAPABILITIES_HEADER, AGENT_PROTOCOL_HEADER, AGENT_PROTOCOL_VERSION, + REQUIRED_AGENT_CAPABILITIES, +}; use tokio::sync::mpsc; const MAX_AGENT_MESSAGE_BYTES: usize = 2 * 1024 * 1024; const MAX_TASK_RESULT_BYTES: usize = 1024 * 1024; -const AGENT_PROTOCOL_VERSION: &str = "1"; -const AGENT_PROTOCOL_HEADER: &str = "x-nodeflare-agent-protocol"; -const AGENT_CAPABILITIES_HEADER: &str = "x-nodeflare-agent-capabilities"; -const REQUIRED_AGENT_CAPABILITIES: [&str; 4] = - ["metrics-v1", "config-v1", "remote-exec-v1", "task-ack-v1"]; #[derive(Deserialize)] #[serde(deny_unknown_fields)] diff --git a/scripts/agent-runtime-test.mjs b/scripts/agent-runtime-test.mjs index f1d69d5..0c56157 100644 --- a/scripts/agent-runtime-test.mjs +++ b/scripts/agent-runtime-test.mjs @@ -2,11 +2,11 @@ import assert from "node:assert/strict"; import { spawn } from "node:child_process"; import { createHash, randomUUID } from "node:crypto"; import { once } from "node:events"; -import { mkdtempSync, rmSync } from "node:fs"; +import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; import { createServer as createHttpServer } from "node:http"; import { createServer as createTcpServer } from "node:net"; import { tmpdir } from "node:os"; -import { join, resolve } from "node:path"; +import { delimiter, join, resolve } from "node:path"; import { setTimeout as delay } from "node:timers/promises"; import { WebSocketServer } from "ws"; import { gunzipSync } from "node:zlib"; @@ -41,7 +41,7 @@ async function waitUntil(predicate, timeout, message) { } } -function startAgent(port, token, name) { +function startAgent(port, token, name, extraEnv = {}) { const proxyUrl = `http://127.0.0.1:${proxy.address().port}`; const child = spawn(binary, ["-e", `http://127.0.0.1:${port}`, "-t", token], { env: { @@ -51,6 +51,7 @@ function startAgent(port, token, name) { HTTPS_PROXY: proxyUrl, https_proxy: proxyUrl, ALL_PROXY: proxyUrl, all_proxy: proxyUrl, NO_PROXY: "", no_proxy: "", + ...extraEnv, }, stdio: ["ignore", "ignore", "pipe"], }); @@ -159,7 +160,23 @@ try { }); }; websocketServer.on("connection", trackTelemetry); - const telemetry = startAgent(upgradeServer.address().port, "runtime-telemetry", "telemetry"); + const telemetryEnv = {}; + if (process.platform === "linux") { + const tools = join(directory, "slow-gpu"); + mkdirSync(tools); + // Align startup near a second boundary, then make a cached GPU refresh cross it. + // CPU/network samples must retain the time at which their counters were read. + writeFileSync(join(tools, "nvidia-smi"), `#!${process.execPath} +const { existsSync, writeFileSync } = require("node:fs"); +const marker = ${JSON.stringify(join(directory, "gpu-initialized"))}; +const initial = !existsSync(marker); +if (initial) writeFileSync(marker, "ready"); +const wait = initial ? (750 - Date.now() % 1000 + 1000) % 1000 : 350; +setTimeout(() => process.stdout.write("10, Runtime test GPU, 256, 1024\\n"), wait); +`, { mode: 0o755 }); + telemetryEnv.PATH = `${tools}${delimiter}${process.env.PATH ?? ""}`; + } + const telemetry = startAgent(upgradeServer.address().port, "runtime-telemetry", "telemetry", telemetryEnv); await waitUntil(() => telemetryConnections.length >= 2 && telemetryConnections[1].commits >= 2, 40_000, "Telemetry reconnect or commit did not complete"); await stopAgent(telemetry); @@ -170,7 +187,7 @@ try { 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] === 1), - "CPU/network samples must be one second apart"); + `CPU/network samples must be one second apart: ${JSON.stringify({ timestamps: before.timestamps, frames: before.frames })}`); 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"); diff --git a/shared/telemetry.rs b/shared/telemetry.rs index 6b8475e..cd194d2 100644 --- a/shared/telemetry.rs +++ b/shared/telemetry.rs @@ -2,6 +2,11 @@ use flate2::{Compression, read::GzDecoder, write::GzEncoder}; use serde::{Deserialize, Serialize, de::DeserializeOwned}; use std::io::{self, Read}; +pub const AGENT_PROTOCOL_VERSION: &str = "1"; +pub const AGENT_PROTOCOL_HEADER: &str = "x-nodeflare-agent-protocol"; +pub const AGENT_CAPABILITIES_HEADER: &str = "x-nodeflare-agent-capabilities"; +pub const REQUIRED_AGENT_CAPABILITIES: [&str; 4] = + ["metrics-v1", "config-v1", "remote-exec-v1", "task-ack-v1"]; pub const MIN_UPLOAD_INTERVAL: u64 = 3; pub const MAX_BATCH_BYTES: usize = 768 * 1024; pub const MAX_DECODED_BYTES: usize = 2 * 1024 * 1024;