From becdb7bb27117c25598e45847d12206c450d29c4 Mon Sep 17 00:00:00 2001 From: imengying Date: Sun, 6 Sep 2026 14:27:11 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96=E5=90=8E=E7=AB=AF=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E5=A4=84=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 6 + agent/src/live_batch.rs | 71 +- agent/src/main.rs | 90 +- .../postgres/02_metric_history_aggregates.sql | 14 + .../sqlite/02_metric_history_aggregates.sql | 14 + backend/src/backup.rs | 65 +- backend/src/db/mod.rs | 169 ++- backend/src/db/queries.rs | 1077 +++++++++++++++-- backend/src/main.rs | 5 + backend/src/models.rs | 4 + backend/src/websocket/agent.rs | 13 +- frontend/src/types.ts | 4 + 12 files changed, 1364 insertions(+), 168 deletions(-) create mode 100644 backend/migrations/postgres/02_metric_history_aggregates.sql create mode 100644 backend/migrations/sqlite/02_metric_history_aggregates.sql diff --git a/README.md b/README.md index fb7bf3d..313b1e2 100644 --- a/README.md +++ b/README.md @@ -125,6 +125,12 @@ database_url = "postgres://nodeflare:password@127.0.0.1:5432/nodeflare?sslmode=d - 恢复建议使用相同版本的 NodeFlare；`pg_dump` 与 SQLite `.backup` 可作为补充。 - 迁移会覆盖目标库已有数据并自动更新 `database_url`，完成后重启生效。 +历史指标按节点设置的保存间隔聚合（默认 60 秒）：CPU、负载、内存和 GPU 保存采样均值，CPU 另存最小/最大值，网速和磁盘 I/O 保留窗口峰值，流量计数保留末值。断网补报按窗口分别写入，重复重传不会重复计数；前台实时状态仍使用最新采样。 + +Agent 保留未确认的本地采样，只有后端事务提交后才清理相应缓存。建议先更新后端再更新 Agent；双方均更新后启用完整的分批补报与聚合，单边更新时保持旧协议兼容。历史表会自动迁移，旧快照按单个采样读取，也可恢复升级前的 ZIP 备份。 + +历史保留天数保持原设置（默认 30 天），指标和延迟记录仍共用该期限。过期清理每分钟运行，每条语句最多处理 1000 条记录，轮流处理各表，每轮最多 50 批并在两秒预算耗尽后停止启动新批次；已开始的语句会执行完成。SQLite 每五分钟尝试小批量增量回收，数据库占用包含主文件、WAL 和 SHM。完整空间回收仍由后台手动触发。 + ## 从源码开发 ```bash diff --git a/agent/src/live_batch.rs b/agent/src/live_batch.rs index 2f70a6f..b31b706 100644 --- a/agent/src/live_batch.rs +++ b/agent/src/live_batch.rs @@ -5,7 +5,7 @@ use super::Report; pub(crate) const MAX_LIVE_BATCH_BYTES: usize = 768 * 1024; pub(crate) fn batch_len(queue: &VecDeque) -> usize { - let mut encoded_bytes = 64_usize; + let mut encoded_bytes = 192_usize; let mut count = 0_usize; for report in queue { let Ok(report_bytes) = serde_json::to_vec(report) else { @@ -26,31 +26,11 @@ pub(crate) fn batch_len(queue: &VecDeque) -> usize { count.min(super::LIVE_QUEUE_CAPACITY) } -pub(crate) fn latest(queue: &VecDeque) -> Vec { - let mut encoded_bytes = 64_usize; - let mut reports = Vec::new(); - for report in queue.iter().rev() { - let Ok(report_bytes) = serde_json::to_vec(report) else { - break; - }; - let next_bytes = encoded_bytes - .saturating_add(report_bytes.len()) - .saturating_add(usize::from(!reports.is_empty())); - if next_bytes > MAX_LIVE_BATCH_BYTES { - break; - } - encoded_bytes = next_bytes; - reports.push(report.clone()); - } - reports.reverse(); - reports -} - #[cfg(test)] mod tests { use std::collections::VecDeque; - use super::{MAX_LIVE_BATCH_BYTES, batch_len, latest}; + use super::{MAX_LIVE_BATCH_BYTES, batch_len}; use crate::Report; #[test] @@ -62,14 +42,16 @@ mod tests { let queue = std::iter::repeat_n(report, 10).collect::>(); let count = batch_len(&queue); assert!((1..10).contains(&count)); - let payload = - crate::live_update_payload(&queue.iter().take(count).cloned().collect::>()) - .unwrap(); + let payload = crate::live_update_payload( + &queue.iter().take(count).cloned().collect::>(), + Some(true), + ) + .unwrap(); assert!(payload.len() <= MAX_LIVE_BATCH_BYTES); } #[test] - fn persistence_batch_keeps_the_newest_samples() { + fn persistence_batch_keeps_the_oldest_unconfirmed_samples() { let queue = (0..10) .map(|timestamp| Report { timestamp, @@ -77,14 +59,47 @@ mod tests { ..Report::default() }) .collect::>(); - let reports = latest(&queue); + let reports = crate::live_persistence_batch(&queue, -1); assert!(!reports.is_empty()); assert!(reports.len() < queue.len()); - assert_eq!(reports.last().unwrap().timestamp, 9); + assert_eq!(reports.first().unwrap().timestamp, 0); assert!( reports .windows(2) .all(|pair| pair[0].timestamp < pair[1].timestamp) ); + let mut persisted = -1; + let mut received = Vec::new(); + loop { + let batch = crate::live_persistence_batch(&queue, persisted); + if batch.is_empty() { + break; + } + let message = crate::live_update_payload(&batch, Some(true)).unwrap(); + assert!(message.len() <= MAX_LIVE_BATCH_BYTES); + persisted = batch.last().unwrap().timestamp; + received.extend(batch.into_iter().map(|report| report.timestamp)); + } + assert_eq!(received, (0..10).collect::>()); + let legacy = crate::legacy_persistence_batch(&queue, -1); + assert_eq!(legacy.last().unwrap().timestamp, 9); + assert!(legacy.len() < queue.len()); + } + + #[test] + fn caps_batch_count_and_preserves_legacy_payload_shape() { + let queue = (0..1000) + .map(|timestamp| Report { + timestamp, + ..Report::default() + }) + .collect::>(); + assert!(batch_len(&queue) <= crate::LIVE_QUEUE_CAPACITY); + let reports = crate::live_persistence_batch(&queue, 100); + assert_eq!(reports.first().unwrap().timestamp, 101); + let message = crate::live_update_payload(&reports, None).unwrap(); + let value: serde_json::Value = serde_json::from_str(&message).unwrap(); + assert!(value.get("persist").is_none()); + assert_eq!(value.as_object().unwrap().len(), 3); } } diff --git a/agent/src/main.rs b/agent/src/main.rs index 9b577a3..514a203 100644 --- a/agent/src/main.rs +++ b/agent/src/main.rs @@ -1746,7 +1746,7 @@ fn report_batch_id(reports: &[Report]) -> String { type LiveSocket = tungstenite::WebSocket>; -fn connect_live(endpoint: &str, token: &str) -> Result<(LiveSocket, Option)> { +fn connect_live(endpoint: &str, token: &str) -> Result<(LiveSocket, Option, bool)> { let mut request = endpoint.into_client_request()?; request .headers_mut() @@ -1770,7 +1770,11 @@ fn connect_live(endpoint: &str, token: &str) -> Result<(LiveSocket, Option) .get("date") .and_then(|value| value.to_str().ok()) .and_then(|value| clock_offset_from_http_date(value, started_ms, ended_ms)); - Ok((socket, clock_offset_ms)) + let persistence_batches = response + .headers() + .get("x-nodeflare-persistence-batches") + .is_some_and(|value| value == "1"); + Ok((socket, clock_offset_ms, persistence_batches)) } fn set_live_read_timeout(socket: &mut LiveSocket, timeout: Option) -> io::Result<()> { @@ -2535,12 +2539,16 @@ fn wait_for_live_ack( } } -fn live_update_payload(reports: &[Report]) -> Result { - Ok(serde_json::to_string(&serde_json::json!({ +fn live_update_payload(reports: &[Report], persist: Option) -> Result { + let mut message = serde_json::json!({ "type": "update", "batchId": report_batch_id(reports), "samples": reports, - }))?) + }); + if let Some(persist) = persist { + message["persist"] = serde_json::json!(persist); + } + Ok(serde_json::to_string(&message)?) } fn observe_persisted_through(target: &AtomicI64, ack: &LiveAck) { @@ -2573,7 +2581,21 @@ fn live_persistence_batch(queue: &VecDeque, timestamp: i64) -> Vec timestamp) .cloned() .collect::>(); - live_batch::latest(&candidates) + let count = live_batch::batch_len(&candidates); + candidates.into_iter().take(count).collect() +} + +fn legacy_persistence_batch(queue: &VecDeque, timestamp: i64) -> Vec { + let candidates = queue + .iter() + .rev() + .filter(|report| report.timestamp > timestamp) + .cloned() + .collect::>(); + let count = live_batch::batch_len(&candidates); + let mut batch = candidates.into_iter().take(count).collect::>(); + batch.reverse(); + batch } fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) { @@ -2586,6 +2608,7 @@ fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) { stats, } = worker; let mut socket: Option = None; + let mut persistence_batches = false; let mut wss_interval = configured_interval .lock() .map_or(Duration::from_secs(1), |interval| *interval); @@ -2602,7 +2625,7 @@ fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) { 'sender: loop { if socket.is_none() { match connect_live(endpoint, token) { - Ok((mut connected, offset_ms)) => { + Ok((mut connected, offset_ms, supports_persistence_batches)) => { observe_clock(&clock, offset_ms); if set_live_read_timeout(&mut connected, Some(LIVE_HINT_READ_TIMEOUT)).is_err() { @@ -2610,6 +2633,7 @@ fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) { continue; } socket = Some(connected); + persistence_batches = supports_persistence_batches; remote_executor.reset_delivery(); accepted_through = persisted_through.load(Ordering::Acquire); next_send_at = Instant::now(); @@ -2636,18 +2660,24 @@ fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) { let Ok(mut queue) = queue_lock.lock() else { return; }; - let batch = loop { + 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_persistence_batch(&queue, persisted); + let persistence_batch = if persistence_batches { + live_persistence_batch(&queue, persisted) + } else { + legacy_persistence_batch(&queue, persisted) + }; if !persistence_batch.is_empty() { - break persistence_batch; + 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; + break (unsent, false, false); } let wake_at = if !unsent.is_empty() { @@ -2732,7 +2762,7 @@ fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) { wss_interval = (*interval).clamp(Duration::from_secs(1), Duration::from_secs(60)); } - let payload = match live_update_payload(&batch) { + let payload = match live_update_payload(&batch, persistence_batches.then_some(persist)) { Ok(payload) => payload, Err(error) => { eprintln!("live payload encode failed: {error}"); @@ -2771,6 +2801,14 @@ fn live_sender_loop(endpoint: &str, token: &str, worker: LiveSenderWorker) { 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) + { + next_probe_at = Instant::now(); + } } } Ok(LiveRead::Config(config)) => { @@ -3769,21 +3807,25 @@ mod tests { #[test] fn encodes_realtime_samples_as_a_batch() { - let payload = live_update_payload(&[ - Report { - timestamp: 10, - cpu: 20.0, - ..Report::default() - }, - Report { - timestamp: 15, - cpu: 30.0, - ..Report::default() - }, - ]) + let payload = live_update_payload( + &[ + Report { + timestamp: 10, + cpu: 20.0, + ..Report::default() + }, + Report { + timestamp: 15, + cpu: 30.0, + ..Report::default() + }, + ], + Some(false), + ) .unwrap(); let value: serde_json::Value = serde_json::from_str(&payload).unwrap(); assert_eq!(value["type"], "update"); + assert_eq!(value["persist"], false); assert_eq!(value["samples"].as_array().unwrap().len(), 2); assert_eq!(value["samples"][1]["cpu"], 30.0); } diff --git a/backend/migrations/postgres/02_metric_history_aggregates.sql b/backend/migrations/postgres/02_metric_history_aggregates.sql new file mode 100644 index 0000000..ca403f9 --- /dev/null +++ b/backend/migrations/postgres/02_metric_history_aggregates.sql @@ -0,0 +1,14 @@ +ALTER TABLE metric_history ADD COLUMN sample_count BIGINT NOT NULL DEFAULT 1 CHECK(sample_count > 0); +ALTER TABLE metric_history ADD COLUMN first_timestamp BIGINT NOT NULL DEFAULT 0; +ALTER TABLE metric_history ADD COLUMN last_timestamp BIGINT NOT NULL DEFAULT 0; +ALTER TABLE metric_history ADD COLUMN cpu_min DOUBLE PRECISION; +ALTER TABLE metric_history ADD COLUMN cpu_max DOUBLE PRECISION; +ALTER TABLE metric_history ADD COLUMN mem_used_max BIGINT; +ALTER TABLE metric_history ADD COLUMN memory_avg DOUBLE PRECISION; +ALTER TABLE metric_history ADD COLUMN memory_min DOUBLE PRECISION; +ALTER TABLE metric_history ADD COLUMN disk_avg DOUBLE PRECISION; +ALTER TABLE metric_history ADD COLUMN disk_min DOUBLE PRECISION; +ALTER TABLE metric_history ADD COLUMN net_in_avg DOUBLE PRECISION; +ALTER TABLE metric_history ADD COLUMN net_in_min DOUBLE PRECISION; +ALTER TABLE metric_history ADD COLUMN net_out_avg DOUBLE PRECISION; +ALTER TABLE metric_history ADD COLUMN net_out_min DOUBLE PRECISION; diff --git a/backend/migrations/sqlite/02_metric_history_aggregates.sql b/backend/migrations/sqlite/02_metric_history_aggregates.sql new file mode 100644 index 0000000..c927f58 --- /dev/null +++ b/backend/migrations/sqlite/02_metric_history_aggregates.sql @@ -0,0 +1,14 @@ +ALTER TABLE metric_history ADD COLUMN sample_count INTEGER NOT NULL DEFAULT 1 CHECK(sample_count > 0); +ALTER TABLE metric_history ADD COLUMN first_timestamp INTEGER NOT NULL DEFAULT 0; +ALTER TABLE metric_history ADD COLUMN last_timestamp INTEGER NOT NULL DEFAULT 0; +ALTER TABLE metric_history ADD COLUMN cpu_min REAL; +ALTER TABLE metric_history ADD COLUMN cpu_max REAL; +ALTER TABLE metric_history ADD COLUMN mem_used_max INTEGER; +ALTER TABLE metric_history ADD COLUMN memory_avg REAL; +ALTER TABLE metric_history ADD COLUMN memory_min REAL; +ALTER TABLE metric_history ADD COLUMN disk_avg REAL; +ALTER TABLE metric_history ADD COLUMN disk_min REAL; +ALTER TABLE metric_history ADD COLUMN net_in_avg REAL; +ALTER TABLE metric_history ADD COLUMN net_in_min REAL; +ALTER TABLE metric_history ADD COLUMN net_out_avg REAL; +ALTER TABLE metric_history ADD COLUMN net_out_min REAL; diff --git a/backend/src/backup.rs b/backend/src/backup.rs index 808a986..d294877 100644 --- a/backend/src/backup.rs +++ b/backend/src/backup.rs @@ -754,9 +754,20 @@ async fn restore_table( bail!("{table} 备份表头无效"); } let header: BackupTableHeader = serde_json::from_str(line.trim_end())?; - if header.table != table || header.columns != columns { + let legacy_history = table == "metric_history" + && columns.len() + == header.columns.len() + crate::db::queries::HISTORY_AGGREGATE_COLUMNS.len() + && columns.starts_with(&header.columns) + && columns[header.columns.len()..] + .iter() + .map(|column| column.name.as_str()) + .eq(crate::db::queries::HISTORY_AGGREGATE_COLUMNS); + if header.table != table || (header.columns != columns && !legacy_history) { bail!("{table} 备份结构与当前数据库不兼容"); } + // Original snapshot archives omit the new aggregate columns; their SQL + // defaults preserve each old row as a single sample. + let columns = &header.columns; let mut restored = 0_usize; let mut batch = Vec::with_capacity(INSERT_BATCH_ROWS); @@ -894,6 +905,58 @@ mod tests { assert_eq!(site_name, "Migrated"); } + #[tokio::test] + async fn restores_legacy_snapshot_archives_and_round_trips_aggregates() { + let source = crate::db::connect("sqlite::memory:").await.unwrap(); + let target = crate::db::connect("sqlite::memory:").await.unwrap(); + let original = sqlx::migrate::Migrator::with_migrations( + crate::db::SQLITE_MIGRATOR + .iter() + .filter(|migration| migration.version == 1) + .cloned() + .collect(), + ); + original.run(source.pool()).await.unwrap(); + target.migrate().await.unwrap(); + sqlx::query( + "INSERT INTO settings(key,value) VALUES \ + ('admin_username','admin'),('admin_password_hash','hash'), \ + ('password_client_salt','salt'),('password_scheme','argon2-client-pbkdf2-v1')", + ) + .execute(source.pool()) + .await + .unwrap(); + sqlx::query("INSERT INTO servers(id,name,token_hash,created_at,updated_at) VALUES ('legacy','Legacy','token',1,1)") + .execute(source.pool()).await.unwrap(); + sqlx::query("INSERT INTO metric_history(server_id,timestamp,cpu,mem_used) VALUES ('legacy',?,75.0,500)") + .bind(crate::db::now()).execute(source.pool()).await.unwrap(); + assert_eq!(copy_database(&source, &target).await.unwrap(), 6); + let points = crate::db::queries::history(&target, "legacy", 1) + .await + .unwrap(); + assert_eq!(points.len(), 1); + assert_eq!(points[0].cpu_min, 75.0); + assert_eq!(points[0].cpu_max, 75.0); + assert_eq!(points[0].mem_used_max, 500); + assert_eq!(points[0].sample_count, 1); + sqlx::query("UPDATE metric_history SET sample_count=4,cpu=50.0,cpu_min=10.0,cpu_max=90.0, \ + memory_avg=50.0,memory_min=10.0,net_in_avg=20.0,net_in_min=1.0,first_timestamp=?,last_timestamp=?") + .bind(crate::db::now() - 3).bind(crate::db::now()).execute(target.pool()).await.unwrap(); + source.migrate().await.unwrap(); + assert_eq!(copy_database(&target, &source).await.unwrap(), 6); + let row = sqlx::query( + "SELECT sample_count,cpu_min,cpu_max,memory_avg,memory_min FROM metric_history", + ) + .fetch_one(source.pool()) + .await + .unwrap(); + assert_eq!(row.try_get::("sample_count").unwrap(), 4); + assert_eq!(row.try_get::("cpu_min").unwrap(), 10.0); + assert_eq!(row.try_get::("cpu_max").unwrap(), 90.0); + assert_eq!(row.try_get::("memory_avg").unwrap(), 50.0); + assert_eq!(row.try_get::("memory_min").unwrap(), 10.0); + } + #[tokio::test] async fn sqlite_backup_round_trip_restores_data_and_clears_sessions() { let db = crate::db::connect("sqlite::memory:").await.unwrap(); diff --git a/backend/src/db/mod.rs b/backend/src/db/mod.rs index c816108..9262573 100644 --- a/backend/src/db/mod.rs +++ b/backend/src/db/mod.rs @@ -10,6 +10,7 @@ use sqlx::sqlite::SqliteConnectOptions; use sqlx::{AnyPool, Row}; use std::collections::HashMap; use std::fs::OpenOptions; +use std::path::PathBuf; use std::str::FromStr; use std::sync::{Mutex, OnceLock}; use std::time::Duration; @@ -18,7 +19,7 @@ pub const SECRET_MASK: &str = "********"; const PASSWORD_SCHEME: &str = "argon2-client-pbkdf2-v1"; const SESSION_ACTIVITY_WRITE_INTERVAL_SECONDS: i64 = 60; -static SQLITE_MIGRATOR: Migrator = sqlx::migrate!("./migrations/sqlite"); +pub(super) static SQLITE_MIGRATOR: Migrator = sqlx::migrate!("./migrations/sqlite"); static POSTGRES_MIGRATOR: Migrator = sqlx::migrate!("./migrations/postgres"); #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -40,6 +41,7 @@ impl DatabaseKind { pub struct Database { pool: AnyPool, kind: DatabaseKind, + sqlite_path: Option, } #[derive(Debug, Clone)] @@ -79,10 +81,11 @@ impl Database { let free_pages = sqlx::query_scalar::<_, i64>("PRAGMA freelist_count") .fetch_one(&self.pool) .await?; - ( - page_count.saturating_mul(page_size), - Some(free_pages.saturating_mul(page_size)), - ) + let size = match self.sqlite_path.as_deref() { + Some(path) => sqlite_storage_size(path).await?, + None => page_count.saturating_mul(page_size), + }; + (size, Some(free_pages.saturating_mul(page_size))) } DatabaseKind::Postgres => ( sqlx::query_scalar::<_, i64>("SELECT pg_database_size(current_database())::BIGINT") @@ -102,13 +105,19 @@ impl Database { pub async fn reclaim_space(&self) -> Result<()> { match self.kind { DatabaseKind::Sqlite => { + let mut connection = self.pool.acquire().await?; sqlx::query("PRAGMA wal_checkpoint(TRUNCATE)") - .execute(&self.pool) + .execute(&mut *connection) + .await?; + sqlx::query("PRAGMA auto_vacuum = INCREMENTAL") + .execute(&mut *connection) + .await?; + sqlx::query("VACUUM").execute(&mut *connection).await?; + sqlx::query("PRAGMA optimize") + .execute(&mut *connection) .await?; - sqlx::query("VACUUM").execute(&self.pool).await?; - sqlx::query("PRAGMA optimize").execute(&self.pool).await?; sqlx::query("PRAGMA wal_checkpoint(TRUNCATE)") - .execute(&self.pool) + .execute(&mut *connection) .await?; } DatabaseKind::Postgres => { @@ -123,13 +132,38 @@ impl Database { pub async fn optimize(&self) -> Result<()> { if self.kind == DatabaseKind::Sqlite { sqlx::query("PRAGMA optimize").execute(&self.pool).await?; - sqlx::query("PRAGMA incremental_vacuum(256)") - .execute(&self.pool) - .await?; } Ok(()) } + pub async fn reclaim_incremental(&self) -> Result<()> { + if self.kind != DatabaseKind::Sqlite { + return Ok(()); + } + let mut connection = self.pool.acquire().await?; + let free_pages = sqlx::query_scalar::<_, i64>("PRAGMA freelist_count") + .fetch_one(&mut *connection) + .await?; + if free_pages == 0 { + return Ok(()); + } + let page_budget = (free_pages / 8).clamp(256, 8192).min(free_pages); + let started = std::time::Instant::now(); + for _ in 0..(page_budget + 127) / 128 { + if started.elapsed() >= Duration::from_millis(200) { + break; + } + sqlx::query("PRAGMA incremental_vacuum(128)") + .execute(&mut *connection) + .await?; + tokio::task::yield_now().await; + } + sqlx::query("PRAGMA wal_checkpoint(PASSIVE)") + .execute(&mut *connection) + .await?; + Ok(()) + } + pub fn sql(&self, statement: &'static str) -> &'static str { if self.kind == DatabaseKind::Sqlite { return statement; @@ -295,7 +329,35 @@ pub async fn connect(database_url: &str) -> Result { }); } let pool = options.connect(database_url).await?; - Ok(Database { pool, kind }) + let sqlite_path = if kind == DatabaseKind::Sqlite { + let databases = sqlx::query("PRAGMA database_list").fetch_all(&pool).await?; + databases.iter().find_map(|row| { + let name: String = row.try_get("name").ok()?; + let file: String = row.try_get("file").ok()?; + (name == "main" && !file.is_empty()).then(|| PathBuf::from(file)) + }) + } else { + None + }; + Ok(Database { + pool, + kind, + sqlite_path, + }) +} + +async fn sqlite_storage_size(path: &std::path::Path) -> Result { + let mut size = 0_i64; + for suffix in ["", "-wal", "-shm"] { + let mut candidate = path.as_os_str().to_os_string(); + candidate.push(suffix); + match tokio::fs::metadata(&candidate).await { + Ok(metadata) => size = size.saturating_add(metadata.len().min(i64::MAX as u64) as i64), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(error) => return Err(error.into()), + } + } + Ok(size) } pub fn database_kind(database_url: &str) -> Result { @@ -904,7 +966,88 @@ mod tests { assert_eq!(stats.kind, "sqlite"); assert!(stats.size_bytes > 0); assert!(stats.reclaimable_bytes.is_some()); + let expected = ["nodeflare.db", "nodeflare.db-wal", "nodeflare.db-shm"] + .into_iter() + .map(|name| { + std::fs::metadata(directory.path().join(name)) + .unwrap() + .len() as i64 + }) + .sum::(); + assert_eq!(stats.size_bytes, expected); + assert!(stats.size_bytes > std::fs::metadata(&path).unwrap().len() as i64); db.reclaim_space().await.unwrap(); + db.pool().close().await; + } + + #[tokio::test] + async fn incremental_reclaim_releases_pages_without_full_vacuum() { + let directory = tempfile::tempdir().unwrap(); + let db = connect(&format!( + "sqlite://{}", + directory.path().join("reclaim.db").display() + )) + .await + .unwrap(); + db.migrate().await.unwrap(); + sqlx::query("CREATE TABLE reclaim_test(id INTEGER PRIMARY KEY, data BLOB)") + .execute(db.pool()) + .await + .unwrap(); + sqlx::query( + "WITH RECURSIVE seq(n) AS (SELECT 1 UNION ALL SELECT n+1 FROM seq WHERE n<3000) \ + INSERT INTO reclaim_test SELECT n,zeroblob(4096) FROM seq", + ) + .execute(db.pool()) + .await + .unwrap(); + sqlx::query("DELETE FROM reclaim_test WHERE id>1") + .execute(db.pool()) + .await + .unwrap(); + let before = sqlx::query_scalar::<_, i64>("PRAGMA freelist_count") + .fetch_one(db.pool()) + .await + .unwrap(); + assert!(before > 256); + db.reclaim_incremental().await.unwrap(); + let after = sqlx::query_scalar::<_, i64>("PRAGMA freelist_count") + .fetch_one(db.pool()) + .await + .unwrap(); + assert!(after < before, "before={before}, after={after}"); + assert!(before - after <= 8192); + assert_eq!( + sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM reclaim_test") + .fetch_one(db.pool()) + .await + .unwrap(), + 1 + ); + db.pool().close().await; + } + + #[tokio::test] + async fn aggregate_migration_preserves_existing_history() { + let db = connect("sqlite::memory:").await.unwrap(); + let original = Migrator::with_migrations( + SQLITE_MIGRATOR + .iter() + .filter(|migration| migration.version == 1) + .cloned() + .collect(), + ); + original.run(db.pool()).await.unwrap(); + sqlx::query("INSERT INTO servers(id,name,token_hash,created_at,updated_at) VALUES ('old','Old','hash',1,1)") + .execute(db.pool()).await.unwrap(); + sqlx::query("INSERT INTO metric_history(server_id,timestamp,cpu,mem_used) VALUES ('old',?,73.0,100)") + .bind(now()).execute(db.pool()).await.unwrap(); + db.migrate().await.unwrap(); + let point = queries::history(&db, "old", 1).await.unwrap().remove(0); + assert_eq!(point.cpu, 73.0); + assert_eq!(point.cpu_min, 73.0); + assert_eq!(point.cpu_max, 73.0); + assert_eq!(point.sample_count, 1); } #[tokio::test] diff --git a/backend/src/db/queries.rs b/backend/src/db/queries.rs index 15f7ab8..ddbde5c 100644 --- a/backend/src/db/queries.rs +++ b/backend/src/db/queries.rs @@ -8,7 +8,7 @@ use crate::models::{ use anyhow::{Context, Result}; use sqlx::any::AnyArguments; use sqlx::{Any, Arguments, AssertSqlSafe, Row, Transaction}; -use std::collections::{HashMap, HashSet}; +use std::collections::{BTreeMap, HashMap, HashSet}; const REMOTE_TASK_ACTIVE_TTL_SECONDS: i64 = 24 * 60 * 60; const AGENT_REPORT_MAX_AGE_SECONDS: i64 = 2 * 60 * 60; @@ -18,6 +18,25 @@ const AGENT_REPORT_MAX_GPUS: usize = 32; const AGENT_REPORT_MAX_LATENCY_RESULTS: usize = 2048; const HISTORY_INSERT_BATCH_ROWS: usize = 500; const LATENCY_INSERT_BATCH_ROWS: usize = 500; +const CLEANUP_DELETE_BATCH_ROWS: i64 = 1000; +const CLEANUP_MAX_PASSES: usize = 10; +const CLEANUP_TIME_BUDGET: std::time::Duration = std::time::Duration::from_secs(2); +pub const HISTORY_AGGREGATE_COLUMNS: [&str; 14] = [ + "sample_count", + "first_timestamp", + "last_timestamp", + "cpu_min", + "cpu_max", + "mem_used_max", + "memory_avg", + "memory_min", + "disk_avg", + "disk_min", + "net_in_avg", + "net_in_min", + "net_out_avg", + "net_out_min", +]; #[derive(Debug, Clone)] pub struct AgentIdentity { @@ -49,6 +68,213 @@ struct TrafficState { used_tx: i64, } +#[derive(Debug, Clone)] +struct HistoryAggregate { + timestamp: i64, + first_timestamp: i64, + latest: AgentReport, + count: f64, + cpu_sum: f64, + cpu_min: f64, + cpu_max: f64, + load1_sum: f64, + load5_sum: f64, + load15_sum: f64, + mem_sum: f64, + mem_max: i64, + swap_sum: f64, + gpu_sum: f64, + net_in_max: f64, + net_out_max: f64, + disk_read_bps_max: f64, + disk_write_bps_max: f64, + disk_read_iops_max: f64, + disk_write_iops_max: f64, + disk_await_max: f64, + disk_utilization_max: f64, + disk_used_max: i64, + processes_max: i64, + tcp_connections_max: i64, + udp_connections_max: i64, + memory_sum: f64, + memory_min: f64, + disk_sum: f64, + disk_min: f64, + net_in_sum: f64, + net_in_min: f64, + net_out_sum: f64, + net_out_min: f64, +} + +impl HistoryAggregate { + fn new(report: &AgentReport, timestamp: i64) -> Self { + Self { + timestamp, + first_timestamp: report.timestamp, + latest: report.clone(), + count: 0.0, + cpu_sum: 0.0, + cpu_min: f64::INFINITY, + cpu_max: f64::NEG_INFINITY, + load1_sum: 0.0, + load5_sum: 0.0, + load15_sum: 0.0, + mem_sum: 0.0, + mem_max: 0, + swap_sum: 0.0, + gpu_sum: 0.0, + net_in_max: 0.0, + net_out_max: 0.0, + disk_read_bps_max: 0.0, + disk_write_bps_max: 0.0, + disk_read_iops_max: 0.0, + disk_write_iops_max: 0.0, + disk_await_max: 0.0, + disk_utilization_max: 0.0, + disk_used_max: 0, + processes_max: 0, + tcp_connections_max: 0, + udp_connections_max: 0, + memory_sum: 0.0, + memory_min: f64::INFINITY, + disk_sum: 0.0, + disk_min: f64::INFINITY, + net_in_sum: 0.0, + net_in_min: f64::INFINITY, + net_out_sum: 0.0, + net_out_min: f64::INFINITY, + } + } + + fn add(&mut self, report: &AgentReport) { + let mem_total = self.latest.mem_total.max(report.mem_total); + let swap_total = self.latest.swap_total.max(report.swap_total); + let disk_total = self.latest.disk_total.max(report.disk_total); + self.latest = report.clone(); + self.latest.mem_total = mem_total; + self.latest.swap_total = swap_total; + self.latest.disk_total = disk_total; + self.count += 1.0; + self.cpu_sum += report.cpu; + self.cpu_min = self.cpu_min.min(report.cpu); + self.cpu_max = self.cpu_max.max(report.cpu); + self.load1_sum += report.load1; + self.load5_sum += report.load5; + self.load15_sum += report.load15; + self.mem_sum += report.mem_used as f64; + self.mem_max = self.mem_max.max(report.mem_used); + self.swap_sum += report.swap_used as f64; + self.gpu_sum += report.gpu_usage; + self.net_in_max = self.net_in_max.max(report.net_in); + self.net_out_max = self.net_out_max.max(report.net_out); + self.disk_read_bps_max = self.disk_read_bps_max.max(report.disk_read_bps); + self.disk_write_bps_max = self.disk_write_bps_max.max(report.disk_write_bps); + self.disk_read_iops_max = self.disk_read_iops_max.max(report.disk_read_iops); + self.disk_write_iops_max = self.disk_write_iops_max.max(report.disk_write_iops); + self.disk_await_max = self.disk_await_max.max(report.disk_await_ms); + self.disk_utilization_max = self.disk_utilization_max.max(report.disk_utilization); + self.disk_used_max = self.disk_used_max.max(report.disk_used); + self.processes_max = self.processes_max.max(report.processes); + self.tcp_connections_max = self.tcp_connections_max.max(report.tcp_connections); + self.udp_connections_max = self.udp_connections_max.max(report.udp_connections); + let memory = capacity_percentage(report.mem_used, report.mem_total); + let disk = capacity_percentage(report.disk_used, report.disk_total); + self.memory_sum += memory; + self.memory_min = self.memory_min.min(memory); + self.disk_sum += disk; + self.disk_min = self.disk_min.min(disk); + self.net_in_sum += report.net_in; + self.net_in_min = self.net_in_min.min(report.net_in); + self.net_out_sum += report.net_out; + self.net_out_min = self.net_out_min.min(report.net_out); + } + + fn finish(mut self) -> HistoryRow { + let count = self.count.max(1.0); + let last_timestamp = self.latest.timestamp; + self.latest.timestamp = self.timestamp; + self.latest.cpu = self.cpu_sum / count; + self.latest.load1 = self.load1_sum / count; + self.latest.load5 = self.load5_sum / count; + self.latest.load15 = self.load15_sum / count; + self.latest.mem_used = (self.mem_sum / count).round() as i64; + self.latest.swap_used = (self.swap_sum / count).round() as i64; + self.latest.net_in = self.net_in_max; + self.latest.net_out = self.net_out_max; + self.latest.gpu_usage = self.gpu_sum / count; + self.latest.disk_read_bps = self.disk_read_bps_max; + self.latest.disk_write_bps = self.disk_write_bps_max; + self.latest.disk_read_iops = self.disk_read_iops_max; + self.latest.disk_write_iops = self.disk_write_iops_max; + self.latest.disk_await_ms = self.disk_await_max; + self.latest.disk_utilization = self.disk_utilization_max; + self.latest.disk_used = self.disk_used_max; + self.latest.processes = self.processes_max; + self.latest.tcp_connections = self.tcp_connections_max; + self.latest.udp_connections = self.udp_connections_max; + HistoryRow { + report: self.latest, + sample_count: self.count as i64, + first_timestamp: self.first_timestamp, + last_timestamp, + cpu_min: self.cpu_min, + cpu_max: self.cpu_max, + mem_used_max: self.mem_max, + memory_avg: self.memory_sum / count, + memory_min: self.memory_min, + disk_avg: self.disk_sum / count, + disk_min: self.disk_min, + net_in_avg: self.net_in_sum / count, + net_in_min: self.net_in_min, + net_out_avg: self.net_out_sum / count, + net_out_min: self.net_out_min, + } + } +} + +#[derive(Debug, Clone)] +struct HistoryRow { + report: AgentReport, + sample_count: i64, + first_timestamp: i64, + last_timestamp: i64, + cpu_min: f64, + cpu_max: f64, + mem_used_max: i64, + memory_avg: f64, + memory_min: f64, + disk_avg: f64, + disk_min: f64, + net_in_avg: f64, + net_in_min: f64, + net_out_avg: f64, + net_out_min: f64, +} + +fn capacity_percentage(used: i64, total: i64) -> f64 { + if total > 0 { + used as f64 * 100.0 / total as f64 + } else { + 0.0 + } +} + +fn aggregate_history(reports: &[AgentReport], interval: i64) -> Vec { + let interval = interval.clamp(15, 3600); + let mut buckets = BTreeMap::::new(); + for report in reports { + let timestamp = report.timestamp.div_euclid(interval) * interval; + buckets + .entry(timestamp) + .or_insert_with(|| HistoryAggregate::new(report, timestamp)) + .add(report); + } + buckets + .into_values() + .map(HistoryAggregate::finish) + .collect() +} + pub async fn list_servers(db: &Database, include_hidden: bool) -> Result> { let statement = if include_hidden { "SELECT s.id, s.name, s.region, s.group_name, s.tags, s.hidden, s.expires_at, \ @@ -357,6 +583,7 @@ pub async fn save_agent_batch( batch_id: &str, reports: &[AgentReport], remote_ip: &str, + persist: Option, ) -> Result { let mut reports = reports.to_vec(); reports.sort_by_key(|report| report.timestamp); @@ -375,7 +602,21 @@ pub async fn save_agent_batch( anyhow::bail!("report batch exceeds 720 samples"); } - let mut transaction = db.pool().begin().await?; + let needs_write = persist != Some(false) + || reports + .iter() + .any(|report| !report.latency_results.is_empty()); + let mut transaction = if db.is_postgres() || !needs_write { + db.pool().begin().await? + } else { + db.pool().begin_with("BEGIN IMMEDIATE").await? + }; + if db.is_postgres() && needs_write { + sqlx::query("SELECT id FROM servers WHERE id=$1 FOR UPDATE") + .bind(&identity.server_id) + .fetch_one(&mut *transaction) + .await?; + } let latest_row = sqlx::query( db.sql("SELECT latest_timestamp, last_batch_id FROM server_latest_state WHERE server_id=?"), ) @@ -415,7 +656,11 @@ pub async fn save_agent_batch( 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); - if persisted_before > 0 && latest.timestamp < next_persist_at { + // Only a complete persistence batch may advance the durable watermark. + // Older agents omit this flag and retain their scheduled persistence behavior. + if persist == Some(false) + || (persist.is_none() && persisted_before > 0 && latest.timestamp < next_persist_at) + { let next_persist_after_ms = next_persist_at.saturating_sub(latest.timestamp).max(1) as u64 * 1000; if reports @@ -435,13 +680,8 @@ pub async fn save_agent_batch( }); } - save_history_rows( - db, - &mut transaction, - &identity.server_id, - std::slice::from_ref(latest), - ) - .await?; + let history_rows = aggregate_history(&reports, report_interval); + save_history_rows(db, &mut transaction, &identity.server_id, &history_rows).await?; save_latency_rows(db, &mut transaction, &identity.server_id, &reports).await?; let latest = reports.last().context("report batch became empty")?; let persisted_through = latest.timestamp; @@ -486,7 +726,7 @@ pub async fn save_agent_batch( .await?; transaction.commit().await?; Ok(PersistResult { - reports: vec![latest.clone()], + reports, persisted: true, persisted_through, next_persist_after_ms: report_interval as u64 * 1000, @@ -677,15 +917,16 @@ async fn save_history_rows( db: &Database, transaction: &mut Transaction<'_, Any>, server_id: &str, - reports: &[AgentReport], + rows: &[HistoryRow], ) -> Result<()> { - for batch in reports.chunks(HISTORY_INSERT_BATCH_ROWS) { + for batch in rows.chunks(HISTORY_INSERT_BATCH_ROWS) { let mut arguments = AnyArguments::default(); let mut parameter_index = 1_usize; let mut value_groups = Vec::with_capacity(batch.len()); - for report in batch { - let mut placeholders = Vec::with_capacity(27); - for _ in 0..27 { + for row in batch { + let report = &row.report; + let mut placeholders = Vec::with_capacity(27 + HISTORY_AGGREGATE_COLUMNS.len()); + for _ in 0..27 + HISTORY_AGGREGATE_COLUMNS.len() { placeholders.push(if db.is_postgres() { let placeholder = format!("${parameter_index}"); parameter_index += 1; @@ -729,26 +970,31 @@ async fn save_history_rows( add!(report.disk_write_iops); add!(report.disk_await_ms); add!(report.disk_utilization); + add!(row.sample_count); + add!(row.first_timestamp); + add!(row.last_timestamp); + add!(row.cpu_min); + add!(row.cpu_max); + add!(row.mem_used_max); + add!(row.memory_avg); + add!(row.memory_min); + add!(row.disk_avg); + add!(row.disk_min); + add!(row.net_in_avg); + add!(row.net_in_min); + add!(row.net_out_avg); + add!(row.net_out_min); } let sql = format!( "INSERT INTO metric_history(server_id, timestamp, cpu, load1, load5, load15, \ mem_used, mem_total, swap_used, swap_total, disk_used, disk_total, net_in, net_out, \ net_rx_total, net_tx_total, uptime, processes, tcp_connections, udp_connections, \ gpu_usage, disk_read_bps, disk_write_bps, disk_read_iops, disk_write_iops, \ - disk_await_ms, disk_utilization) VALUES {} \ - ON CONFLICT(server_id, timestamp) DO UPDATE SET cpu=excluded.cpu, \ - load1=excluded.load1, load5=excluded.load5, load15=excluded.load15, \ - mem_used=excluded.mem_used, mem_total=excluded.mem_total, \ - swap_used=excluded.swap_used, swap_total=excluded.swap_total, \ - disk_used=excluded.disk_used, disk_total=excluded.disk_total, net_in=excluded.net_in, \ - net_out=excluded.net_out, net_rx_total=excluded.net_rx_total, \ - net_tx_total=excluded.net_tx_total, uptime=excluded.uptime, \ - processes=excluded.processes, tcp_connections=excluded.tcp_connections, \ - udp_connections=excluded.udp_connections, gpu_usage=excluded.gpu_usage, \ - disk_read_bps=excluded.disk_read_bps, disk_write_bps=excluded.disk_write_bps, \ - disk_read_iops=excluded.disk_read_iops, disk_write_iops=excluded.disk_write_iops, \ - disk_await_ms=excluded.disk_await_ms, disk_utilization=excluded.disk_utilization", - value_groups.join(",") + disk_await_ms, disk_utilization, {}) VALUES {} \ + ON CONFLICT(server_id, timestamp) DO UPDATE SET {}", + HISTORY_AGGREGATE_COLUMNS.join(","), + value_groups.join(","), + history_merge_sql(), ); sqlx::query_with(AssertSqlSafe(sql), arguments) .execute(&mut **transaction) @@ -757,6 +1003,99 @@ async fn save_history_rows( Ok(()) } +fn history_merge_sql() -> String { + let mut updates = Vec::new(); + for column in [ + "cpu", + "load1", + "load5", + "load15", + "gpu_usage", + "mem_used", + "swap_used", + ] { + let mean = format!( + "(CAST(metric_history.{column} AS DOUBLE PRECISION)*metric_history.sample_count \ + +excluded.{column}*excluded.sample_count)/(metric_history.sample_count+excluded.sample_count)" + ); + let value = if matches!(column, "mem_used" | "swap_used") { + format!("CAST(ROUND({mean}) AS BIGINT)") + } else { + mean + }; + updates.push(format!("{column}={value}")); + } + for column in [ + "mem_total", + "swap_total", + "disk_used", + "disk_total", + "net_in", + "net_out", + "processes", + "tcp_connections", + "udp_connections", + "disk_read_bps", + "disk_write_bps", + "disk_read_iops", + "disk_write_iops", + "disk_await_ms", + "disk_utilization", + ] { + updates.push(format!( + "{column}=CASE WHEN metric_history.{column}>excluded.{column} \ + THEN metric_history.{column} ELSE excluded.{column} END" + )); + } + for (column, fallback, minimum) in [ + ("cpu_min", "cpu", true), + ("cpu_max", "cpu", false), + ("mem_used_max", "mem_used", false), + ] { + let old = format!("COALESCE(metric_history.{column},metric_history.{fallback})"); + let comparison = if minimum { "<" } else { ">" }; + updates.push(format!( + "{column}=CASE WHEN {old}{comparison}excluded.{column} \ + THEN {old} ELSE excluded.{column} END" + )); + } + for (prefix, fallback) in [ + ( + "memory", + "CASE WHEN metric_history.mem_total>0 THEN \ + CAST(metric_history.mem_used AS DOUBLE PRECISION)*100.0/metric_history.mem_total ELSE 0.0 END", + ), + ( + "disk", + "CASE WHEN metric_history.disk_total>0 THEN \ + CAST(metric_history.disk_used AS DOUBLE PRECISION)*100.0/metric_history.disk_total ELSE 0.0 END", + ), + ("net_in", "metric_history.net_in"), + ("net_out", "metric_history.net_out"), + ] { + let mean = format!("COALESCE(metric_history.{prefix}_avg,{fallback})"); + let min = format!("COALESCE(metric_history.{prefix}_min,{fallback})"); + updates.push(format!("{prefix}_avg=({mean}*metric_history.sample_count \ + +excluded.{prefix}_avg*excluded.sample_count)/(metric_history.sample_count+excluded.sample_count)")); + updates.push(format!( + "{prefix}_min=CASE WHEN {min}, @@ -827,22 +1166,28 @@ pub async fn history(db: &Database, server_id: &str, hours: i64) -> Result=? \ + FROM metric_history WHERE server_id=? \ + AND COALESCE(NULLIF(last_timestamp,0),timestamp)>=? \ ) samples GROUP BY bucket_timestamp ORDER BY bucket_timestamp", )) .bind(bucket) @@ -856,10 +1201,14 @@ pub async fn history(db: &Database, server_id: &str, hours: i64) -> Result("cpu")?, + cpu_min: row.try_get("cpu_min")?, + cpu_max: row.try_get("cpu_max")?, + sample_count: row.try_get("sample_count")?, load1: row.try_get::("load1")?, load5: row.try_get::("load5")?, load15: row.try_get::("load15")?, mem_used: row.try_get::("mem_used")?.round() as i64, + mem_used_max: row.try_get("mem_used_max")?, mem_total: row.try_get("mem_total")?, swap_used: row.try_get::("swap_used")?.round() as i64, swap_total: row.try_get("swap_total")?, @@ -886,39 +1235,74 @@ pub async fn history(db: &Database, server_id: &str, hours: i64) -> Result Result<()> { + cleanup_database_with_budget(db, retention_days, CLEANUP_MAX_PASSES, CLEANUP_TIME_BUDGET).await +} + +async fn cleanup_database_with_budget( + db: &Database, + retention_days: i64, + max_passes: usize, + budget: std::time::Duration, +) -> Result<()> { let current = now(); let cutoff = current - retention_days.clamp(1, 3650) * 86_400; let active_task_cutoff = current - REMOTE_TASK_ACTIVE_TTL_SECONDS; - let mut transaction = db.pool().begin().await?; - sqlx::query(db.sql( - "UPDATE remote_tasks SET status='failed', completed_at=?, \ - result=CASE WHEN result='' THEN '任务等待超过 24 小时，已自动取消' ELSE result END \ - WHERE status IN ('pending','sent') AND requested_at= budget { + return Ok(()); + } + if *finished { + continue; + } + let affected = if index == 0 { + sqlx::query(db.sql( + "UPDATE remote_tasks SET status='failed', completed_at=?, \ + result=CASE WHEN result='' THEN '任务等待超过 24 小时，已自动取消' ELSE result END \ + WHERE status IN ('pending','sent') AND id IN ( \ + SELECT id FROM remote_tasks WHERE status IN ('pending','sent') AND requested_at Result> { let rules = list_alert_rules(db).await?; + let report_interval = + sqlx::query_scalar::<_, i64>(db.sql("SELECT report_interval FROM servers WHERE id=?")) + .bind(server_id) + .fetch_optional(db.pool()) + .await? + .unwrap_or(60); let mut evaluations = Vec::new(); for rule in rules.into_iter().filter(|rule| { rule.enabled && (rule.server_ids.is_empty() || rule.server_ids.iter().any(|id| id == server_id)) }) { - let expression = match rule.metric.as_str() { - "cpu" => "cpu", - "memory" => { - "CASE WHEN mem_total>0 THEN CAST(mem_used AS DOUBLE PRECISION) * \ - CAST(100 AS DOUBLE PRECISION) / CAST(mem_total AS DOUBLE PRECISION) \ - ELSE CAST(0 AS DOUBLE PRECISION) END" - } - "disk" => { - "CASE WHEN disk_total>0 THEN CAST(disk_used AS DOUBLE PRECISION) * \ - CAST(100 AS DOUBLE PRECISION) / CAST(disk_total AS DOUBLE PRECISION) \ - ELSE CAST(0 AS DOUBLE PRECISION) END" - } - "net_in" => "net_in / CAST(1048576 AS DOUBLE PRECISION)", - "net_out" => "net_out / CAST(1048576 AS DOUBLE PRECISION)", + let (average_expression, minimum_expression) = match rule.metric.as_str() { + "cpu" => ("cpu", "COALESCE(cpu_min, cpu)"), + "memory" => ( + "COALESCE(memory_avg, CASE WHEN mem_total>0 THEN CAST(mem_used AS DOUBLE PRECISION)*100.0/mem_total ELSE 0.0 END)", + "COALESCE(memory_min, CASE WHEN mem_total>0 THEN CAST(mem_used AS DOUBLE PRECISION)*100.0/mem_total ELSE 0.0 END)", + ), + "disk" => ( + "COALESCE(disk_avg, CASE WHEN disk_total>0 THEN CAST(disk_used AS DOUBLE PRECISION)*100.0/disk_total ELSE 0.0 END)", + "COALESCE(disk_min, CASE WHEN disk_total>0 THEN CAST(disk_used AS DOUBLE PRECISION)*100.0/disk_total ELSE 0.0 END)", + ), + "net_in" => ( + "COALESCE(net_in_avg, net_in)/1048576.0", + "COALESCE(net_in_min, net_in)/1048576.0", + ), + "net_out" => ( + "COALESCE(net_out_avg, net_out)/1048576.0", + "COALESCE(net_out_min, net_out)/1048576.0", + ), _ => continue, }; - let query = if db.is_postgres() { - format!( - "SELECT MIN(timestamp) AS first_timestamp, AVG({expression}) AS average_value, \ - MIN({expression}) AS minimum_value, COUNT(*) AS sample_count \ - FROM metric_history WHERE server_id=$1 AND timestamp>=$2" - ) - } else { - format!( - "SELECT MIN(timestamp) AS first_timestamp, AVG({expression}) AS average_value, \ - MIN({expression}) AS minimum_value, COUNT(*) AS sample_count \ - FROM metric_history WHERE server_id=? AND timestamp>=?" - ) - }; - let since = now() - rule.duration_minutes * 60; + let filter = db.sql( + "server_id=? AND timestamp>=? AND COALESCE(NULLIF(last_timestamp,0),timestamp)>=?", + ); + let query = format!( + "SELECT MIN(COALESCE(NULLIF(first_timestamp,0),timestamp)) AS first_timestamp, \ + MAX(COALESCE(NULLIF(last_timestamp,0),timestamp)) AS last_timestamp, \ + SUM(({average_expression}) * sample_count) / NULLIF(SUM(sample_count), 0) AS average_value, \ + MIN({minimum_expression}) AS minimum_value, COUNT(*) AS sample_count \ + FROM metric_history WHERE {filter}" + ); + let current = now(); + let since = current - rule.duration_minutes * 60; let row = sqlx::query(sqlx::AssertSqlSafe(query.as_str())) .bind(server_id) + .bind(since - 3600) .bind(since) .fetch_one(db.pool()) .await?; let first: Option = row.try_get("first_timestamp")?; + let last: Option = row.try_get("last_timestamp")?; let count: i64 = row.try_get("sample_count")?; let average: Option = row.try_get("average_value")?; let minimum: Option = row.try_get("minimum_value")?; @@ -1401,7 +1795,9 @@ pub async fn evaluate_resource_rules( } else { average.unwrap_or_default() }; - let covered = count > 0 && first.is_some_and(|first| first <= since + 5); + let covered = count > 0 + && first.is_some_and(|first| first <= since + 5) + && last.is_some_and(|last| last >= current - report_interval - 5); evaluations.push(ResourceAlertEvaluation { triggered: covered && value >= rule.threshold, rule, @@ -1984,6 +2380,327 @@ mod tests { } } + async fn test_server() -> (Database, AgentIdentity) { + let db = crate::db::connect("sqlite::memory:").await.unwrap(); + db.migrate().await.unwrap(); + let (_, token) = create_server(&db, &server_input(0)).await.unwrap(); + let identity = agent_identity(&db, &token).await.unwrap().unwrap(); + (db, identity) + } + + fn sample(timestamp: i64, cpu: f64) -> AgentReport { + AgentReport { + timestamp, + cpu, + load1: cpu / 10.0, + mem_used: cpu as i64 * 10, + mem_total: 1000, + disk_used: cpu as i64 * 10, + disk_total: 1000, + net_in: cpu * 1_048_576.0, + net_out: cpu * 2_097_152.0, + disk_read_bps: cpu * 100.0, + gpu_usage: cpu, + net_rx_total: timestamp * 10, + net_tx_total: timestamp * 20, + ..AgentReport::default() + } + } + + #[tokio::test] + async fn aggregate_batches_merge_weighted_values_and_ignore_retries() { + let (db, identity) = test_server().await; + let start = now().div_euclid(60) * 60 - 120; + let reports = [ + sample(start, 10.0), + sample(start + 1, 90.0), + sample(start + 2, 20.0), + ]; + let first = save_agent_batch(&db, &identity, "first", &reports[..2], "", Some(true)) + .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..], "", Some(true)) + .await + .unwrap(); + assert_eq!(second.persisted_through, start + 2); + for id in ["overlap", "another-retry"] { + let retry = save_agent_batch(&db, &identity, id, &reports, "", Some(true)) + .await + .unwrap(); + assert!(retry.reports.is_empty()); + assert_eq!(retry.persisted_through, start + 2); + } + let points = history(&db, &identity.server_id, 1).await.unwrap(); + assert_eq!(points.len(), 1); + let point = &points[0]; + assert_eq!(point.sample_count, 3); + assert_eq!(point.cpu, 40.0); + assert_eq!(point.cpu_min, 10.0); + assert_eq!(point.cpu_max, 90.0); + assert_eq!(point.mem_used, 400); + assert_eq!(point.mem_used_max, 900); + assert_eq!(point.net_in, 90.0 * 1_048_576.0); + assert_eq!(point.disk_read_bps, 9000.0); + assert_eq!(point.net_rx_total, reports[2].net_rx_total); + let latest = list_servers(&db, true).await.unwrap().remove(0); + assert_eq!(latest.timestamp, Some(start + 2)); + assert_eq!(latest.cpu, Some(20.0)); + } + + #[tokio::test] + async fn realtime_batches_cannot_acknowledge_unpersisted_history() { + let (db, identity) = test_server().await; + let start = now() - 180; + let reports = (0..121) + .map(|offset| sample(start + offset, 25.0)) + .collect::>(); + save_agent_batch(&db, &identity, "initial", &reports[..1], "", Some(true)) + .await + .unwrap(); + let live = save_agent_batch(&db, &identity, "live", &reports[120..], "", Some(false)) + .await + .unwrap(); + assert!(!live.persisted); + assert_eq!(live.persisted_through, 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, + "", + Some(true), + ) + .await + .unwrap(); + assert!(result.persisted); + assert_eq!(result.persisted_through, batch.last().unwrap().timestamp); + } + let count = sqlx::query_scalar::<_, i64>( + "SELECT CAST(SUM(sample_count) AS BIGINT) FROM metric_history", + ) + .fetch_one(db.pool()) + .await + .unwrap(); + assert_eq!(count, 121); + } + + #[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)], + "", + Some(true), + ) + .await + .unwrap(); + sqlx::query( + "CREATE TRIGGER reject_latest BEFORE UPDATE ON server_latest_state \ + BEGIN SELECT RAISE(ABORT, 'test write failure'); END", + ) + .execute(db.pool()) + .await + .unwrap(); + let next = sample(start + 1, 90.0); + assert!( + save_agent_batch( + &db, + &identity, + "next", + std::slice::from_ref(&next), + "", + Some(true) + ) + .await + .is_err() + ); + let count = sqlx::query_scalar::<_, i64>( + "SELECT CAST(SUM(sample_count) AS BIGINT) FROM metric_history", + ) + .fetch_one(db.pool()) + .await + .unwrap(); + assert_eq!(count, 1); + let timestamp = + sqlx::query_scalar::<_, i64>("SELECT latest_timestamp FROM server_latest_state") + .fetch_one(db.pool()) + .await + .unwrap(); + assert_eq!(timestamp, start); + sqlx::query("DROP TRIGGER reject_latest") + .execute(db.pool()) + .await + .unwrap(); + let result = save_agent_batch(&db, &identity, "next", &[next], "", Some(true)) + .await + .unwrap(); + assert_eq!(result.persisted_through, start + 1); + } + + #[tokio::test] + async fn history_query_weights_different_windows_and_reads_legacy_snapshots() { + let (db, identity) = test_server().await; + let start = (now() - 1200).div_euclid(120) * 120; + let reports = [ + sample(start, 90.0), + sample(start + 1, 90.0), + sample(start + 2, 90.0), + sample(start + 60, 10.0), + ]; + save_agent_batch(&db, &identity, "weighted", &reports, "", Some(true)) + .await + .unwrap(); + let points = history(&db, &identity.server_id, 24).await.unwrap(); + assert_eq!(points.len(), 1); + assert_eq!(points[0].cpu, 70.0); + assert_eq!(points[0].cpu_min, 10.0); + assert_eq!(points[0].cpu_max, 90.0); + assert_eq!(points[0].sample_count, 4); + sqlx::query( + "INSERT INTO metric_history(server_id,timestamp,cpu,mem_used) VALUES (?,?,33.0,44)", + ) + .bind(&identity.server_id) + .bind(start + 240) + .execute(db.pool()) + .await + .unwrap(); + let points = history(&db, &identity.server_id, 24).await.unwrap(); + let legacy = points.last().unwrap(); + assert_eq!(legacy.cpu_min, 33.0); + assert_eq!(legacy.cpu_max, 33.0); + assert_eq!(legacy.mem_used_max, 44); + assert_eq!(legacy.sample_count, 1); + } + + #[tokio::test] + async fn history_includes_a_bucket_crossing_the_query_start() { + let (db, identity) = test_server().await; + let current = now(); + let since = current - 3600; + sqlx::query( + "INSERT INTO metric_history(server_id,timestamp,cpu,first_timestamp,last_timestamp) \ + VALUES (?,?,?,?,?)", + ) + .bind(&identity.server_id) + .bind(since - 4) + .bind(42.0) + .bind(since - 4) + .bind(since + 10) + .execute(db.pool()) + .await + .unwrap(); + + let points = history(&db, &identity.server_id, 1).await.unwrap(); + assert!(points.iter().any(|point| point.cpu == 42.0)); + } + + #[tokio::test] + async fn aggregate_alerts_use_average_and_minimum_not_bucket_peaks() { + let (db, identity) = test_server().await; + let start = now() - 120; + 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, "", Some(true)) + .await + .unwrap(); + for metric in ["cpu", "memory", "disk", "net_in", "net_out"] { + for aggregation in ["average", "continuous"] { + create_alert_rule( + &db, + &AlertRuleInput { + name: format!("{metric}-{aggregation}"), + metric: metric.to_string(), + threshold: 50.0, + duration_minutes: 2, + aggregation: aggregation.to_string(), + enabled: true, + server_ids: vec![identity.server_id.clone()], + }, + ) + .await + .unwrap(); + } + } + let evaluations = evaluate_resource_rules(&db, &identity.server_id) + .await + .unwrap(); + assert_eq!(evaluations.len(), 10); + for evaluation in evaluations { + assert_eq!( + evaluation.triggered, + evaluation.rule.aggregation == "average", + "{}: {}", + evaluation.rule.name, + evaluation.value + ); + } + } + + #[tokio::test] + async fn concurrent_sqlite_retries_are_counted_once() { + let directory = tempfile::tempdir().unwrap(); + let db = crate::db::connect(&format!( + "sqlite://{}", + directory.path().join("concurrent.db").display() + )) + .await + .unwrap(); + db.migrate().await.unwrap(); + let (_, token) = create_server(&db, &server_input(0)).await.unwrap(); + let identity = agent_identity(&db, &token).await.unwrap().unwrap(); + let reports = vec![sample(now() - 60, 20.0)]; + let mut handles = Vec::new(); + for index in 0..8 { + let db = db.clone(); + let identity = identity.clone(); + let reports = reports.clone(); + handles.push(tokio::spawn(async move { + save_agent_batch( + &db, + &identity, + &format!("retry-{index}"), + &reports, + "", + Some(true), + ) + .await + })); + } + for handle in handles { + handle.await.unwrap().unwrap(); + } + let count = sqlx::query_scalar::<_, i64>( + "SELECT CAST(SUM(sample_count) AS BIGINT) FROM metric_history", + ) + .fetch_one(db.pool()) + .await + .unwrap(); + assert_eq!(count, 1); + db.pool().close().await; + } + + #[test] + fn aggregation_uses_the_configured_window_and_preserves_counter_resets() { + let reports = [sample(3600, 10.0), sample(3614, 30.0), sample(3615, 90.0)]; + let rows = aggregate_history(&reports, 15); + assert_eq!(rows.len(), 2); + assert_eq!(rows[0].report.cpu, 20.0); + assert_eq!(rows[0].sample_count, 2); + assert_eq!(aggregate_history(&reports, 3600).len(), 1); + let mut reports = reports; + reports[2].net_rx_total = 5; + assert_eq!(aggregate_history(&reports, 60)[0].report.net_rx_total, 5); + } + #[test] fn validates_agent_report_bounds() { let current = now(); @@ -2065,7 +2782,7 @@ mod tests { .await .unwrap(); let identity = agent_identity(&db, &token).await.unwrap().unwrap(); - let start = now() - 600; + let start = now().div_euclid(60) * 60 - 600; let reports = (0..600) .map(|offset| AgentReport { timestamp: start + offset, @@ -2084,7 +2801,7 @@ mod tests { }) .collect::>(); - let result = save_agent_batch(&db, &identity, "large-batch", &reports, "127.0.0.1") + let result = save_agent_batch(&db, &identity, "large-batch", &reports, "127.0.0.1", None) .await .unwrap(); let rows = sqlx::query_scalar::<_, i64>( @@ -2102,8 +2819,8 @@ mod tests { .await .unwrap(); assert!(result.persisted); - assert_eq!(result.reports.len(), 1); - assert_eq!(rows, 1); + assert_eq!(result.reports.len(), 600); + assert_eq!(rows, 10); assert_eq!(latency_rows, 600); let persisted_at = start + 599; @@ -2116,6 +2833,7 @@ mod tests { ..AgentReport::default() }], "127.0.0.1", + None, ) .await .unwrap(); @@ -2132,6 +2850,7 @@ mod tests { ..AgentReport::default() }], "127.0.0.1", + None, ) .await .unwrap(); @@ -2143,7 +2862,7 @@ mod tests { .fetch_one(db.pool()) .await .unwrap(); - assert_eq!(rows, 2); + assert_eq!(rows, 11); } #[tokio::test] @@ -2182,6 +2901,7 @@ mod tests { ..AgentReport::default() }], "127.0.0.1", + None, ) .await .unwrap(); @@ -2256,6 +2976,163 @@ mod tests { assert_eq!(active_count, 1); } + #[tokio::test] + 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)], + "", + Some(true), + ) + .await + .unwrap(); + let task_id = create_latency_task( + &db, + &LatencyTaskInput { + name: "Cleanup".to_string(), + task_type: "tcp".to_string(), + target: "example.com".to_string(), + port: Some(443), + interval_seconds: 60, + default_enabled: false, + server_ids: vec![identity.server_id.clone()], + }, + ) + .await + .unwrap(); + let old = current - 3 * 86_400; + sqlx::query( + "WITH RECURSIVE seq(n) AS (SELECT 1 UNION ALL SELECT n+1 FROM seq WHERE n<2100) \ + INSERT INTO metric_history(server_id,timestamp) SELECT ?,?+n FROM seq", + ) + .bind(&identity.server_id) + .bind(old) + .execute(db.pool()) + .await + .unwrap(); + sqlx::query( + "INSERT INTO latency_results(task_id,server_id,timestamp,latency_ms,packet_loss) \ + SELECT ?,server_id,timestamp,10.0,0.0 FROM metric_history", + ) + .bind(&task_id) + .execute(db.pool()) + .await + .unwrap(); + sqlx::query("INSERT INTO alert_states(state_key,active,updated_at) VALUES ('active',1,?),('expired',0,?)") + .bind(old).bind(old).execute(db.pool()).await.unwrap(); + cleanup_database_with_budget(&db, 1, 1, std::time::Duration::ZERO) + .await + .unwrap(); + assert_eq!( + sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM metric_history") + .fetch_one(db.pool()) + .await + .unwrap(), + 2101 + ); + cleanup_database_with_budget(&db, 1, 1, std::time::Duration::from_secs(60)) + .await + .unwrap(); + for table in ["metric_history", "latency_results"] { + let count = sqlx::query_scalar::<_, i64>(AssertSqlSafe(format!( + "SELECT COUNT(*) FROM {table}" + ))) + .fetch_one(db.pool()) + .await + .unwrap(); + assert_eq!(count, 1101, "{table}"); + } + // Later tables still get a turn while metrics have a backlog. + assert_eq!( + sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM alert_states WHERE active=0") + .fetch_one(db.pool()) + .await + .unwrap(), + 0 + ); + cleanup_database(&db, 1).await.unwrap(); + for table in [ + "metric_history", + "latency_results", + "servers", + "server_latest_state", + "server_traffic_state", + "latency_task_servers", + "alert_states", + ] { + let count = sqlx::query_scalar::<_, i64>(AssertSqlSafe(format!( + "SELECT COUNT(*) FROM {table}" + ))) + .fetch_one(db.pool()) + .await + .unwrap(); + assert_eq!(count, 1, "{table}"); + } + delete_server(&db, &identity.server_id).await.unwrap(); + for table in [ + "metric_history", + "latency_results", + "servers", + "server_latest_state", + "server_traffic_state", + "latency_task_servers", + ] { + let count = sqlx::query_scalar::<_, i64>(AssertSqlSafe(format!( + "SELECT COUNT(*) FROM {table}" + ))) + .fetch_one(db.pool()) + .await + .unwrap(); + assert_eq!(count, 0, "{table}"); + } + } + + #[tokio::test] + async fn cleanup_bounds_expired_task_updates_and_keeps_live_tasks() { + let (db, identity) = test_server().await; + let old = now() - 3 * 86_400; + sqlx::query( + "WITH RECURSIVE seq(n) AS (SELECT 1 UNION ALL SELECT n+1 FROM seq WHERE n<1100) \ + INSERT INTO remote_tasks(id,server_id,command,requested_by,requested_at) \ + SELECT CAST(n AS TEXT),?,'uptime','admin',? FROM seq", + ) + .bind(&identity.server_id) + .bind(old) + .execute(db.pool()) + .await + .unwrap(); + let live = create_remote_task(&db, &identity.server_id, "uptime", "admin") + .await + .unwrap(); + cleanup_database_with_budget(&db, 1, 1, std::time::Duration::from_secs(60)) + .await + .unwrap(); + assert_eq!( + sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM remote_tasks WHERE status='failed'") + .fetch_one(db.pool()) + .await + .unwrap(), + 1000 + ); + assert_eq!( + sqlx::query_scalar::<_, i64>( + "SELECT COUNT(*) FROM remote_tasks WHERE status='pending'" + ) + .fetch_one(db.pool()) + .await + .unwrap(), + 101 + ); + assert_eq!( + remote_task(&db, &live.id).await.unwrap().unwrap().status, + "pending" + ); + } + #[tokio::test] async fn remote_task_delivery_and_results_are_idempotent() { let db = super::super::connect("sqlite::memory:").await.unwrap(); diff --git a/backend/src/main.rs b/backend/src/main.rs index 8dc56ff..cdbb606 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -408,6 +408,11 @@ fn spawn_maintenance(state: Arc) { { tracing::error!(%error, "database optimization failed"); } + if maintenance_runs.is_multiple_of(5) + && let Err(error) = state.db.reclaim_incremental().await + { + tracing::error!(%error, "incremental database reclaim failed"); + } } if let Err(error) = exchange::refresh(&state.db, &state.http, false).await { tracing::warn!(%error, "scheduled exchange-rate refresh failed"); diff --git a/backend/src/models.rs b/backend/src/models.rs index edadf72..9f51f30 100644 --- a/backend/src/models.rs +++ b/backend/src/models.rs @@ -286,10 +286,14 @@ pub struct ServerView { pub struct HistoryPoint { pub timestamp: i64, pub cpu: f64, + pub cpu_min: f64, + pub cpu_max: f64, + pub sample_count: i64, pub load1: f64, pub load5: f64, pub load15: f64, pub mem_used: i64, + pub mem_used_max: i64, pub mem_total: i64, pub swap_used: i64, pub swap_total: i64, diff --git a/backend/src/websocket/agent.rs b/backend/src/websocket/agent.rs index b10ea3c..4c70b3c 100644 --- a/backend/src/websocket/agent.rs +++ b/backend/src/websocket/agent.rs @@ -30,6 +30,8 @@ struct UpdateMessage { #[serde(rename = "batchId")] batch_id: String, samples: Vec, + #[serde(default)] + persist: Option, } #[derive(Deserialize)] @@ -86,9 +88,15 @@ pub async fn handle( }; let remote_ip = client_ip(&headers, peer, &state.config.trusted_proxies); let token = token.to_string(); - ws.max_message_size(MAX_AGENT_MESSAGE_BYTES) + let mut response = ws + .max_message_size(MAX_AGENT_MESSAGE_BYTES) .max_frame_size(MAX_AGENT_MESSAGE_BYTES) - .on_upgrade(move |socket| run(socket, state, identity, remote_ip, token)) + .on_upgrade(move |socket| run(socket, state, identity, remote_ip, token)); + response.headers_mut().insert( + "x-nodeflare-persistence-batches", + axum::http::HeaderValue::from_static("1"), + ); + response } async fn run( @@ -233,6 +241,7 @@ async fn handle_text( &update.batch_id, &update.samples, remote_ip, + update.persist, ) .await { diff --git a/frontend/src/types.ts b/frontend/src/types.ts index 44b7b9b..1aa25ab 100644 --- a/frontend/src/types.ts +++ b/frontend/src/types.ts @@ -221,10 +221,14 @@ export interface AdminServer extends ServerSummary { export interface HistoryPoint { timestamp: number; cpu: number; + cpu_min?: number; + cpu_max?: number; + sample_count?: number; load1: number; load5: number; load15: number; mem_used: number; + mem_used_max?: number; mem_total: number; swap_used: number; swap_total: number;