From 467961eedfca468c1f0ba0414f3a2712439afed6 Mon Sep 17 00:00:00 2001 From: imengying Date: Tue, 18 Aug 2026 11:31:42 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96=E6=95=B0=E6=8D=AE=E5=BA=93?= =?UTF-8?q?=E8=B0=83=E7=94=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- agent/agent.sh | 5 +- migrations/initial_01.sql | 2 + scripts/smoke-test.sh | 35 ++++-- src/auth.rs | 10 +- src/db.rs | 249 +++++++++++++++++++++++++++++--------- src/latency.rs | 19 ++- src/live.rs | 42 ++++--- src/routes/agent.rs | 5 +- wrangler.jsonc | 2 +- 9 files changed, 273 insertions(+), 96 deletions(-) diff --git a/agent/agent.sh b/agent/agent.sh index dc5acae..e147940 100644 --- a/agent/agent.sh +++ b/agent/agent.sh @@ -219,7 +219,8 @@ install_agent() { "ReadWritePaths=$INSTALL_DIR" \ '' \ '[Install]' \ - 'WantedBy=multi-user.target' > "$SERVICE_FILE" + 'WantedBy=multi-user.target' > "$SERVICE_FILE" + chmod 600 "$SERVICE_FILE" systemctl daemon-reload systemctl enable --now "$SERVICE_NAME" systemctl is-active --quiet "$SERVICE_NAME" || { @@ -236,7 +237,7 @@ install_agent() { "supervisor=\"supervise-daemon\"" \ "respawn_delay=10" \ 'depend() { need net; }' > "$OPENRC_FILE" - chmod 755 "$OPENRC_FILE" + chmod 700 "$OPENRC_FILE" rc-update add "$SERVICE_NAME" default rc-service "$SERVICE_NAME" restart rc-service "$SERVICE_NAME" status >/dev/null || { diff --git a/migrations/initial_01.sql b/migrations/initial_01.sql index c205e38..e4fa88b 100644 --- a/migrations/initial_01.sql +++ b/migrations/initial_01.sql @@ -63,6 +63,7 @@ CREATE TABLE metric_history ( disk_write_iops REAL NOT NULL DEFAULT 0, disk_await_ms REAL NOT NULL DEFAULT 0, disk_utilization REAL NOT NULL DEFAULT 0, + sample_count INTEGER NOT NULL DEFAULT 1 CHECK(sample_count > 0), latest_timestamp INTEGER NOT NULL, latest_json TEXT NOT NULL CHECK(json_valid(latest_json)), latency_json TEXT NOT NULL DEFAULT '[]' CHECK(json_valid(latency_json)), @@ -96,6 +97,7 @@ CREATE TABLE metric_history_hourly ( disk_write_iops REAL NOT NULL DEFAULT 0, disk_await_ms REAL NOT NULL DEFAULT 0, disk_utilization REAL NOT NULL DEFAULT 0, + sample_count INTEGER NOT NULL DEFAULT 1 CHECK(sample_count > 0), latest_timestamp INTEGER NOT NULL, latest_json TEXT NOT NULL CHECK(json_valid(latest_json)), latency_json TEXT NOT NULL DEFAULT '[]' CHECK(json_valid(latency_json)), diff --git a/scripts/smoke-test.sh b/scripts/smoke-test.sh index 22c521b..6fb430e 100644 --- a/scripts/smoke-test.sh +++ b/scripts/smoke-test.sh @@ -118,7 +118,7 @@ request -H "Authorization: Bearer $admin_token" \ agent_version=$(sh scripts/resolve-version.sh) sample=$(jq -nc --arg agent_version "$agent_version" '{ - timestamp:(now|floor),cpu:18.5,load1:0.42,load5:0.36,load15:0.31, + timestamp:((((now|floor) / 60 | floor) + 1) * 60 + 10),cpu:18.5,load1:0.42,load5:0.36,load15:0.31, mem_used:2147483648,mem_total:4294967296,swap_used:0,swap_total:0,disk_used:21474836480,disk_total:53687091200, net_in:4096,net_out:2048,net_rx_total:1073741824,net_tx_total:536870912, uptime:86400,processes:90,tcp_connections:18,udp_connections:4,cpu_cores:2, @@ -147,23 +147,40 @@ request -H "Authorization: Bearer $agent_token" \ request -H "Authorization: Bearer $admin_token" "$MONITOR_BASE_URL/api/admin/servers" | \ jq -e --arg id "$server_id" '.servers | any(.id == $id and .last_ip == "8.8.8.8")' >/dev/null -latency_report=$(printf '%s' "$report" | jq --arg task_id "$latency_task_id" '.samples[0].timestamp=(now|floor) | .samples[0].latency_results=[{task_id:$task_id,timestamp:(now|floor),latency_ms:28.4,packet_loss:25}]') +latency_report=$(printf '%s' "$report" | jq --arg task_id "$latency_task_id" \ + '.samples[0].timestamp += 1 | + .samples[0].latency_results=[{ + task_id:$task_id, + timestamp:.samples[0].timestamp, + latency_ms:28.4, + packet_loss:25 + }]') request -H "Authorization: Bearer $agent_token" -H 'Content-Type: application/json' \ --data "$latency_report" "$MONITOR_BASE_URL/api/agent/report" | jq -e '.collect_interval == 5' >/dev/null +latency_report_second=$(printf '%s' "$report" | jq --arg task_id "$latency_task_id" \ + '.samples[0].timestamp += 2 | + .samples[0].latency_results=[{ + task_id:$task_id, + timestamp:.samples[0].timestamp, + latency_ms:48.4, + packet_loss:75 + }]') +request -H "Authorization: Bearer $agent_token" -H 'Content-Type: application/json' \ + --data "$latency_report_second" "$MONITOR_BASE_URL/api/agent/report" | jq -e '.collect_interval == 5' >/dev/null # A whole Agent batch must collapse into one traffic-cycle write while preserving # counter resets. Replaying the same batch must not count any bytes twice. traffic_report=$(printf '%s' "$report" | jq '.samples = [ - (.samples[0] | .timestamp += 2 | .net_rx_total=2147483648 | .net_tx_total=1073741824), - (.samples[0] | .timestamp += 3 | .net_rx_total=268435456 | .net_tx_total=134217728), - (.samples[0] | .timestamp += 4 | .net_rx_total=536870912 | .net_tx_total=268435456) + (.samples[0] | .timestamp += 3 | .net_rx_total=2147483648 | .net_tx_total=1073741824), + (.samples[0] | .timestamp += 4 | .net_rx_total=268435456 | .net_tx_total=134217728), + (.samples[0] | .timestamp += 5 | .net_rx_total=536870912 | .net_tx_total=268435456) ]') request -H "Authorization: Bearer $agent_token" -H 'Content-Type: application/json' \ --data "$traffic_report" "$MONITOR_BASE_URL/api/agent/report" | jq -e '.collect_interval == 5' >/dev/null request -H "Authorization: Bearer $agent_token" -H 'Content-Type: application/json' \ --data "$traffic_report" "$MONITOR_BASE_URL/api/agent/report" | jq -e '.collect_interval == 5' >/dev/null -request -H "Authorization: Bearer $admin_token" "$MONITOR_BASE_URL/api/bootstrap" | jq -e --arg id "$server_id" --arg task_id "$latency_task_id" '.servers | any(.id == $id and .cpu == 18.5 and .gpu_usage == 32.5 and .disk_await_ms == 1.4 and (.gpus | length) == 1 and (.disks | length) == 1 and .disk_used == 21474836480 and .traffic_limit == 107374182400 and .net_rx_total == 2684354560 and .net_tx_total == 1342177280 and .price == 9.9 and (has("last_ip") | not) and (.latency | any(.task_id == $task_id and .latency_ms == 28.4 and .packet_loss == 25)))' >/dev/null +request -H "Authorization: Bearer $admin_token" "$MONITOR_BASE_URL/api/bootstrap" | jq -e --arg id "$server_id" --arg task_id "$latency_task_id" '.servers | any(.id == $id and .cpu == 18.5 and .gpu_usage == 32.5 and .disk_await_ms == 1.4 and (.gpus | length) == 1 and (.disks | length) == 1 and .disk_used == 21474836480 and .traffic_limit == 107374182400 and .net_rx_total == 2684354560 and .net_tx_total == 1342177280 and .price == 9.9 and (has("last_ip") | not) and (.latency | any(.task_id == $task_id and .latency_ms == 48.4 and .packet_loss == 75)))' >/dev/null # Changing the reset day starts a new traffic cycle. Only growth after the first # sample in that cycle counts toward usage. @@ -171,8 +188,8 @@ request -H "Authorization: Bearer $admin_token" -H 'Content-Type: application/js --data "$(printf '%s' "$server_input" | jq '.reset_day=2')" \ "$MONITOR_BASE_URL/api/admin/servers/$server_id" >/dev/null cycle_report=$(printf '%s' "$report" | jq '.samples = [ - (.samples[0] | .timestamp += 5 | .net_rx_total=805306368 | .net_tx_total=536870912), - (.samples[0] | .timestamp += 6 | .net_rx_total=1073741824 | .net_tx_total=805306368) + (.samples[0] | .timestamp += 6 | .net_rx_total=805306368 | .net_tx_total=536870912), + (.samples[0] | .timestamp += 7 | .net_rx_total=1073741824 | .net_tx_total=805306368) ]') request -H "Authorization: Bearer $agent_token" -H 'Content-Type: application/json' \ --data "$cycle_report" "$MONITOR_BASE_URL/api/agent/report" | jq -e '.collect_interval == 5' >/dev/null @@ -196,7 +213,7 @@ history_cache_header=$(monitor_curl --silent --dump-header - --output /dev/null -H "Authorization: Bearer $admin_token" "$MONITOR_BASE_URL/api/history/$server_id?hours=1" | \ tr -d '\r' | awk -F ': ' 'tolower($1) == "x-cache" { print $2 }' | tail -1) [ "$history_cache_header" = "HIT" ] -request -H "Authorization: Bearer $admin_token" "$MONITOR_BASE_URL/api/latency/$server_id?hours=1" | jq -e --arg task_id "$latency_task_id" '(.tasks | any(.id == $task_id)) and (.points | any(.task_id == $task_id and .latency_ms == 28.4))' >/dev/null +request -H "Authorization: Bearer $admin_token" "$MONITOR_BASE_URL/api/latency/$server_id?hours=1" | jq -e --arg task_id "$latency_task_id" '(.tasks | any(.id == $task_id)) and (.points | any(.task_id == $task_id and .latency_ms > 38.399 and .latency_ms < 38.401 and .packet_loss > 49.999 and .packet_loss < 50.001))' >/dev/null # An Agent may have already measured a task when the administrator removes its # assignment. The stale result must not block delivery of the new task list. diff --git a/src/auth.rs b/src/auth.rs index 845e16b..d10ac4b 100644 --- a/src/auth.rs +++ b/src/auth.rs @@ -30,12 +30,10 @@ pub fn sha256_hex(value: &str) -> String { } fn secure_eq(left: &str, right: &str) -> bool { - if left.len() != right.len() { - return false; - } - left.as_bytes() - .iter() - .zip(right.as_bytes()) + let left = Sha256::digest(left.as_bytes()); + let right = Sha256::digest(right.as_bytes()); + left.iter() + .zip(right.iter()) .fold(0_u8, |diff, (a, b)| diff | (a ^ b)) == 0 } diff --git a/src/db.rs b/src/db.rs index 69c8a93..5af9635 100644 --- a/src/db.rs +++ b/src/db.rs @@ -302,6 +302,10 @@ impl HistoryMetricAggregate { } } + pub fn sample_count(&self) -> i64 { + self.count.min(i64::MAX as u64) as i64 + } + pub fn point(&self) -> Option { if self.count == 0 { return None; @@ -983,13 +987,15 @@ pub async fn history(db: &D1Database, id: &str, hours: i64) -> Result Result Result Result<()> { - let mut aggregate = HistoryMetricAggregate::default(); - aggregate.extend(reports); - let Some(history) = aggregate.point() else { - return Ok(()); - }; - let mut latency = crate::latency::LatencyMetricAggregates::default(); - let received_at = now(); - for report in reports { - latency.extend(&report.latency_results, received_at); - } +pub(crate) fn reports_after(reports: &[AgentReport], timestamp: i64) -> Vec { + reports + .iter() + .filter(|report| report.timestamp > timestamp) + .cloned() + .collect() +} + +pub async fn save_reports( + db: &D1Database, + server_id: &str, + reports: &[AgentReport], +) -> Result> { let context: TrafficCounterContext = db .prepare( r#"WITH latest_state AS ( @@ -1056,8 +1064,33 @@ pub async fn save_reports(db: &D1Database, server_id: &str, reports: &[AgentRepo .await? .ok_or_else(|| worker::Error::RustError("节点不存在".to_string()))?; let (mut traffic, configured_reset_day) = context.into_parts(); - traffic.extend(reports, configured_reset_day); - save_reports_with_history(db, server_id, reports, &history, &traffic, &latency).await + let fresh_reports = reports_after(reports, traffic.timestamp); + if fresh_reports.is_empty() { + return Ok(None); + } + + let mut aggregate = HistoryMetricAggregate::default(); + aggregate.extend(&fresh_reports); + let Some(history) = aggregate.point() else { + return Ok(None); + }; + let mut latency = crate::latency::LatencyMetricAggregates::default(); + let received_at = now(); + for report in &fresh_reports { + latency.extend(&report.latency_results, received_at); + } + traffic.extend(&fresh_reports, configured_reset_day); + save_reports_with_history( + db, + server_id, + &fresh_reports, + &history, + aggregate.sample_count(), + &traffic, + &latency, + ) + .await?; + Ok(Some(history)) } pub async fn save_reports_with_history( @@ -1065,6 +1098,7 @@ pub async fn save_reports_with_history( server_id: &str, reports: &[AgentReport], history_point: &HistoryPoint, + sample_count: i64, traffic: &TrafficCounterState, latency: &crate::latency::LatencyMetricAggregates, ) -> Result<()> { @@ -1089,27 +1123,49 @@ pub async fn save_reports_with_history( net_in, net_out, net_rx_total, net_tx_total, processes, tcp_connections, udp_connections, gpu_usage, disk_read_bps, disk_write_bps, disk_read_iops, disk_write_iops, - disk_await_ms, disk_utilization, latest_timestamp, latest_json, latency_json + disk_await_ms, disk_utilization, sample_count, + latest_timestamp, latest_json, latency_json ) VALUES ( ?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18, ?19, ?20, ?21, ?22, ?23, ?24, ?25, ?26, - ?27, ?28, ?29 + ?27, ?28, ?29, ?30 ) 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, + cpu=(metric_history.cpu * metric_history.sample_count + + excluded.cpu * excluded.sample_count) / + (metric_history.sample_count + excluded.sample_count), + load1=(metric_history.load1 * metric_history.sample_count + + excluded.load1 * excluded.sample_count) / + (metric_history.sample_count + excluded.sample_count), + load5=(metric_history.load5 * metric_history.sample_count + + excluded.load5 * excluded.sample_count) / + (metric_history.sample_count + excluded.sample_count), + load15=(metric_history.load15 * metric_history.sample_count + + excluded.load15 * excluded.sample_count) / + (metric_history.sample_count + excluded.sample_count), + mem_used=CAST((metric_history.mem_used * metric_history.sample_count + + excluded.mem_used * excluded.sample_count) / + (metric_history.sample_count + excluded.sample_count) AS INTEGER), + mem_total=excluded.mem_total, + swap_used=CAST((metric_history.swap_used * metric_history.sample_count + + excluded.swap_used * excluded.sample_count) / + (metric_history.sample_count + excluded.sample_count) AS INTEGER), 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, 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, + disk_total=excluded.disk_total, + net_in=MAX(metric_history.net_in, excluded.net_in), + net_out=MAX(metric_history.net_out, excluded.net_out), + net_rx_total=excluded.net_rx_total, + net_tx_total=excluded.net_tx_total, + processes=MAX(metric_history.processes, excluded.processes), + tcp_connections=MAX(metric_history.tcp_connections, excluded.tcp_connections), + udp_connections=MAX(metric_history.udp_connections, excluded.udp_connections), + gpu_usage=excluded.gpu_usage, + disk_read_bps=MAX(metric_history.disk_read_bps, excluded.disk_read_bps), + disk_write_bps=MAX(metric_history.disk_write_bps, excluded.disk_write_bps), + disk_read_iops=MAX(metric_history.disk_read_iops, excluded.disk_read_iops), + disk_write_iops=MAX(metric_history.disk_write_iops, excluded.disk_write_iops), + disk_await_ms=MAX(metric_history.disk_await_ms, excluded.disk_await_ms), + disk_utilization=MAX(metric_history.disk_utilization, excluded.disk_utilization), + sample_count=metric_history.sample_count + excluded.sample_count, latest_timestamp=MAX(metric_history.latest_timestamp, excluded.latest_timestamp), latest_json=CASE WHEN excluded.latest_timestamp >= metric_history.latest_timestamp THEN excluded.latest_json @@ -1117,17 +1173,57 @@ pub async fn save_reports_with_history( latency_json=CASE WHEN json_array_length(excluded.latency_json) = 0 THEN metric_history.latency_json ELSE ( - SELECT json_group_array(json(item)) FROM ( - SELECT value AS item FROM json_each(excluded.latency_json) + WITH latency_rows AS ( + SELECT + json_extract(value, '$.task_id') AS task_id, + CAST(json_extract(value, '$.timestamp') AS INTEGER) AS timestamp, + CAST(json_extract(value, '$.latency_ms') AS REAL) AS latency_ms, + CAST(json_extract(value, '$.packet_loss') AS REAL) AS packet_loss, + CAST(json_extract(value, '$.sample_count') AS INTEGER) AS sample_count, + CAST(json_extract(value, '$.success_count') AS INTEGER) AS success_count, + CAST(json_extract(value, '$.latest_timestamp') AS INTEGER) AS latest_timestamp, + CAST(json_extract(value, '$.latest_latency_ms') AS REAL) AS latest_latency_ms, + CAST(json_extract(value, '$.latest_packet_loss') AS REAL) AS latest_packet_loss + FROM json_each(metric_history.latency_json) UNION ALL - SELECT previous.value AS item FROM json_each(metric_history.latency_json) previous - WHERE NOT EXISTS ( - SELECT 1 FROM json_each(excluded.latency_json) current - WHERE json_extract(current.value, '$.task_id') = - json_extract(previous.value, '$.task_id') - ) - ) - ) END"#, + SELECT + json_extract(value, '$.task_id'), + CAST(json_extract(value, '$.timestamp') AS INTEGER), + CAST(json_extract(value, '$.latency_ms') AS REAL), + CAST(json_extract(value, '$.packet_loss') AS REAL), + CAST(json_extract(value, '$.sample_count') AS INTEGER), + CAST(json_extract(value, '$.success_count') AS INTEGER), + CAST(json_extract(value, '$.latest_timestamp') AS INTEGER), + CAST(json_extract(value, '$.latest_latency_ms') AS REAL), + CAST(json_extract(value, '$.latest_packet_loss') AS REAL) + FROM json_each(excluded.latency_json) + ), latency_ranked AS ( + SELECT *, ROW_NUMBER() OVER ( + PARTITION BY task_id ORDER BY latest_timestamp DESC + ) AS latest_position + FROM latency_rows + ), latency_tasks AS ( + SELECT task_id, MAX(timestamp) AS timestamp, + CASE WHEN SUM(success_count) > 0 + THEN SUM(latency_ms * success_count) / SUM(success_count) + ELSE -1 END AS latency_ms, + SUM(packet_loss * sample_count) / SUM(sample_count) AS packet_loss, + SUM(sample_count) AS sample_count, + SUM(success_count) AS success_count, + MAX(CASE WHEN latest_position = 1 THEN latest_timestamp END) AS latest_timestamp, + MAX(CASE WHEN latest_position = 1 THEN latest_latency_ms END) AS latest_latency_ms, + MAX(CASE WHEN latest_position = 1 THEN latest_packet_loss END) AS latest_packet_loss + FROM latency_ranked GROUP BY task_id + ) SELECT json_group_array(json_object( + 'task_id', task_id, 'timestamp', timestamp, + 'latency_ms', latency_ms, 'packet_loss', packet_loss, + 'sample_count', sample_count, 'success_count', success_count, + 'latest_timestamp', latest_timestamp, + 'latest_latency_ms', latest_latency_ms, + 'latest_packet_loss', latest_packet_loss + )) FROM (SELECT * FROM latency_tasks ORDER BY task_id) + ) END + WHERE excluded.latest_timestamp > metric_history.latest_timestamp"#, ) .bind(&[ text(server_id), @@ -1144,8 +1240,8 @@ pub async fn save_reports_with_history( number(history_point.disk_total), number(history_point.net_in), number(history_point.net_out), - number(history_point.net_rx_total), - number(history_point.net_tx_total), + number(traffic.used_rx), + number(traffic.used_tx), number(history_point.processes), number(history_point.tcp_connections), number(history_point.udp_connections), @@ -1156,6 +1252,7 @@ pub async fn save_reports_with_history( number(history_point.disk_write_iops), number(history_point.disk_await_ms), number(history_point.disk_utilization), + number(sample_count.max(1)), number(latest_timestamp), text(&latest_json), text(&latency_json), @@ -1563,21 +1660,28 @@ pub async fn cleanup_history(db: &D1Database, retention_days: i64) -> Result<()> ), metrics AS ( SELECT server_id, bucket, - AVG(cpu) AS cpu, AVG(load1) AS load1, AVG(load5) AS load5, - AVG(load15) AS load15, CAST(AVG(mem_used) AS INTEGER) AS mem_used, - MAX(mem_total) AS mem_total, CAST(AVG(swap_used) AS INTEGER) AS swap_used, - MAX(swap_total) AS swap_total, CAST(AVG(disk_used) AS INTEGER) AS disk_used, + SUM(cpu * sample_count) / SUM(sample_count) AS cpu, + SUM(load1 * sample_count) / SUM(sample_count) AS load1, + SUM(load5 * sample_count) / SUM(sample_count) AS load5, + SUM(load15 * sample_count) / SUM(sample_count) AS load15, + CAST(SUM(mem_used * sample_count) / SUM(sample_count) AS INTEGER) AS mem_used, + MAX(mem_total) AS mem_total, + CAST(SUM(swap_used * sample_count) / SUM(sample_count) AS INTEGER) AS swap_used, + MAX(swap_total) AS swap_total, + CAST(SUM(disk_used * sample_count) / SUM(sample_count) AS INTEGER) AS disk_used, MAX(disk_total) AS disk_total, MAX(net_in) AS net_in, MAX(net_out) AS net_out, MAX(net_rx_total) AS net_rx_total, MAX(net_tx_total) AS net_tx_total, CAST(MAX(processes) AS INTEGER) AS processes, CAST(MAX(tcp_connections) AS INTEGER) AS tcp_connections, CAST(MAX(udp_connections) AS INTEGER) AS udp_connections, - AVG(gpu_usage) AS gpu_usage, MAX(disk_read_bps) AS disk_read_bps, + SUM(gpu_usage * sample_count) / SUM(sample_count) AS gpu_usage, + MAX(disk_read_bps) AS disk_read_bps, MAX(disk_write_bps) AS disk_write_bps, MAX(disk_read_iops) AS disk_read_iops, MAX(disk_write_iops) AS disk_write_iops, MAX(disk_await_ms) AS disk_await_ms, MAX(disk_utilization) AS disk_utilization, + SUM(sample_count) AS sample_count, MAX(CASE WHEN latest_position = 1 THEN latest_timestamp END) AS latest_timestamp, MAX(CASE WHEN latest_position = 1 THEN latest_json END) AS latest_json FROM ranked GROUP BY server_id, bucket @@ -1586,6 +1690,8 @@ pub async fn cleanup_history(db: &D1Database, retention_days: i64) -> Result<()> json_extract(j.value, '$.task_id') AS task_id, CAST(json_extract(j.value, '$.latency_ms') AS REAL) AS latency_ms, CAST(json_extract(j.value, '$.packet_loss') AS REAL) AS packet_loss, + CAST(json_extract(j.value, '$.sample_count') AS INTEGER) AS sample_count, + CAST(json_extract(j.value, '$.success_count') AS INTEGER) AS success_count, CAST(json_extract(j.value, '$.latest_timestamp') AS INTEGER) AS latest_timestamp, CAST(json_extract(j.value, '$.latest_latency_ms') AS REAL) AS latest_latency_ms, CAST(json_extract(j.value, '$.latest_packet_loss') AS REAL) AS latest_packet_loss, @@ -1597,9 +1703,12 @@ pub async fn cleanup_history(db: &D1Database, retention_days: i64) -> Result<()> WHERE h.timestamp >= ?1 AND h.timestamp < ?2 ), latency_tasks AS ( SELECT server_id, bucket, task_id, - CASE WHEN SUM(CASE WHEN latency_ms >= 0 THEN 1 ELSE 0 END) > 0 - THEN AVG(CASE WHEN latency_ms >= 0 THEN latency_ms END) ELSE -1 END AS latency_ms, - AVG(packet_loss) AS packet_loss, + CASE WHEN SUM(success_count) > 0 + THEN SUM(latency_ms * success_count) / SUM(success_count) + ELSE -1 END AS latency_ms, + SUM(packet_loss * sample_count) / SUM(sample_count) AS packet_loss, + SUM(sample_count) AS sample_count, + SUM(success_count) AS success_count, MAX(CASE WHEN latest_position = 1 THEN latest_timestamp END) AS latest_timestamp, MAX(CASE WHEN latest_position = 1 THEN latest_latency_ms END) AS latest_latency_ms, MAX(CASE WHEN latest_position = 1 THEN latest_packet_loss END) AS latest_packet_loss @@ -1608,6 +1717,7 @@ pub async fn cleanup_history(db: &D1Database, retention_days: i64) -> Result<()> SELECT server_id, bucket, json_group_array(json_object( 'task_id', task_id, 'timestamp', bucket, 'latency_ms', latency_ms, 'packet_loss', packet_loss, + 'sample_count', sample_count, 'success_count', success_count, 'latest_timestamp', latest_timestamp, 'latest_latency_ms', latest_latency_ms, 'latest_packet_loss', latest_packet_loss @@ -1618,14 +1728,15 @@ pub async fn cleanup_history(db: &D1Database, retention_days: i64) -> Result<()> swap_used, swap_total, disk_used, disk_total, net_in, net_out, net_rx_total, net_tx_total, processes, tcp_connections, udp_connections, gpu_usage, disk_read_bps, disk_write_bps, disk_read_iops, disk_write_iops, - disk_await_ms, disk_utilization, latest_timestamp, latest_json, latency_json + disk_await_ms, disk_utilization, sample_count, + latest_timestamp, latest_json, latency_json ) SELECT m.server_id, m.bucket, m.cpu, m.load1, m.load5, m.load15, m.mem_used, m.mem_total, m.swap_used, m.swap_total, m.disk_used, m.disk_total, m.net_in, m.net_out, m.net_rx_total, m.net_tx_total, m.processes, m.tcp_connections, m.udp_connections, m.gpu_usage, m.disk_read_bps, m.disk_write_bps, m.disk_read_iops, m.disk_write_iops, m.disk_await_ms, - m.disk_utilization, m.latest_timestamp, m.latest_json, + m.disk_utilization, m.sample_count, m.latest_timestamp, m.latest_json, COALESCE(l.latency_json, '[]') FROM metrics m LEFT JOIN latency_packed l ON l.server_id = m.server_id AND l.bucket = m.bucket @@ -1642,6 +1753,7 @@ pub async fn cleanup_history(db: &D1Database, retention_days: i64) -> Result<()> 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, + sample_count=excluded.sample_count, latest_timestamp=excluded.latest_timestamp, latest_json=excluded.latest_json, latency_json=excluded.latency_json"#, ) @@ -2010,6 +2122,7 @@ mod tests { aggregate.extend(&reports[1..]); let point = aggregate.point().unwrap(); + assert_eq!(aggregate.sample_count(), 2); assert_eq!(point.timestamp, 120); assert_eq!(point.cpu, 20.0); assert_eq!(point.mem_used, 200); @@ -2028,4 +2141,26 @@ mod tests { assert_eq!(point.net_rx_total, 2_000); assert_eq!(point.net_tx_total, 3_000); } + + #[test] + fn excludes_reports_already_persisted() { + let reports = [ + AgentReport { + timestamp: 99, + ..AgentReport::default() + }, + AgentReport { + timestamp: 100, + ..AgentReport::default() + }, + AgentReport { + timestamp: 101, + ..AgentReport::default() + }, + ]; + + let fresh = super::reports_after(&reports, 100); + assert_eq!(fresh.len(), 1); + assert_eq!(fresh[0].timestamp, 101); + } } diff --git a/src/latency.rs b/src/latency.rs index 3c65eb3..91f7f9f 100644 --- a/src/latency.rs +++ b/src/latency.rs @@ -31,6 +31,8 @@ pub(crate) struct StoredLatencyResult { timestamp: i64, latency_ms: f64, packet_loss: f64, + sample_count: u64, + success_count: u64, latest_timestamp: i64, latest_latency_ms: f64, latest_packet_loss: f64, @@ -104,6 +106,8 @@ impl LatencyMetricAggregates { aggregate.2 / aggregate.1 as f64 }, packet_loss: aggregate.3 / aggregate.0.max(1) as f64, + sample_count: aggregate.0, + success_count: aggregate.1, latest_timestamp: aggregate.4, latest_latency_ms: aggregate.5, latest_packet_loss: aggregate.6, @@ -392,6 +396,8 @@ pub async fn history(db: &D1Database, server_id: &str, hours: i64) -> Result Result Result= 0 THEN 1 ELSE 0 END) > 0 \ - THEN AVG(CASE WHEN h.latency_ms >= 0 THEN h.latency_ms END) ELSE -1 END AS latency_ms, \ - AVG(h.packet_loss) AS packet_loss \ + CASE WHEN SUM(h.success_count) > 0 \ + THEN SUM(h.latency_ms * h.success_count) / SUM(h.success_count) \ + ELSE -1 END AS latency_ms, \ + SUM(h.packet_loss * h.sample_count) / SUM(h.sample_count) AS packet_loss \ FROM ({source}) h INNER JOIN latency_tasks t ON t.id = h.task_id \ INNER JOIN latency_task_servers a \ ON a.task_id = h.task_id AND a.server_id = h.server_id \ @@ -479,6 +490,8 @@ mod tests { assert_eq!(history[0].timestamp, 120); assert_eq!(history[0].latency_ms, 15.0); assert_eq!(history[0].packet_loss, 12.5); + assert_eq!(history[0].sample_count, 2); + assert_eq!(history[0].success_count, 2); assert_eq!(history[0].latest_timestamp, 149); assert_eq!(history[0].latest_latency_ms, 20.0); assert_eq!(history[0].latest_packet_loss, 25.0); diff --git a/src/live.rs b/src/live.rs index 12c0fa3..980e152 100644 --- a/src/live.rs +++ b/src/live.rs @@ -6,7 +6,7 @@ use worker::*; use crate::db::{AgentLiveContext, AlertMetricRow, HistoryMetricAggregate}; use crate::latency::LatencyMetricAggregates; -use crate::models::{AgentLatencyResult, AgentReport, AlertRuleView, HistoryPoint, ServerView}; +use crate::models::{AgentReport, AlertRuleView, HistoryPoint, ServerView}; const MAX_LIVE_SAMPLES: usize = 720; const MAX_LIVE_LATENCY_RESULTS: usize = 4096; @@ -856,7 +856,7 @@ impl DurableObject for LiveHub { return ws.close(Some(1007), Some("invalid live metric batch")); } let received_at = crate::now(); - let mut latency_results = Vec::::new(); + let mut latency_result_count = 0usize; for report in &batch.samples { if report.timestamp <= 0 || (report.timestamp - received_at).abs() > 7200 @@ -864,9 +864,10 @@ impl DurableObject for LiveHub { { return ws.close(Some(1007), Some("invalid live metric sample")); } - latency_results.extend(report.latency_results.iter().cloned()); + latency_result_count = + latency_result_count.saturating_add(report.latency_results.len()); } - if latency_results.len() > MAX_LIVE_LATENCY_RESULTS { + if latency_result_count > MAX_LIVE_LATENCY_RESULTS { return ws.close(Some(1009), Some("too many live latency results")); } let Some(server_id) = attachment.server_id.clone() else { @@ -876,8 +877,14 @@ impl DurableObject for LiveHub { .traffic .take() .ok_or_else(|| Error::RustError("missing live traffic state".to_string()))?; + let previous_traffic_timestamp = traffic.timestamp; + let fresh_reports = crate::db::reports_after(&batch.samples, previous_traffic_timestamp); + let latency_results = fresh_reports + .iter() + .flat_map(|report| report.latency_results.iter().cloned()) + .collect::>(); let (payload, mut cached_samples) = - batch_update_parts_at(&server_id, &batch.samples, &mut traffic, received_at)?; + batch_update_parts_at(&server_id, &fresh_reports, &mut traffic, received_at)?; trim_cached_live_samples(&mut cached_samples); let traffic_state = crate::db::TrafficCounterState { cycle_key: traffic.cycle_key, @@ -889,17 +896,22 @@ impl DurableObject for LiveHub { used_tx: traffic.used_tx, }; attachment.traffic = Some(traffic); - attachment.latest_samples = cached_samples; - attachment.latest_received_at = received_at; - attachment.history_aggregate.extend(&batch.samples); - attachment - .latency_aggregates - .extend(&latency_results, received_at); + if !fresh_reports.is_empty() { + attachment.latest_samples = cached_samples; + attachment.latest_received_at = received_at; + attachment.history_aggregate.extend(&fresh_reports); + attachment + .latency_aggregates + .extend(&latency_results, received_at); + } let report_interval = attachment.report_interval.clamp(15, 3600); let due_for_d1 = received_at.saturating_sub(attachment.last_d1_write_at) >= report_interval; let flush_active = self.active_d1_flushes.borrow().contains(&server_id); - let mut flush = if (due_for_d1 || attachment.d1_flush.is_some()) && !flush_active { + let mut flush = if !fresh_reports.is_empty() + && (due_for_d1 || attachment.d1_flush.is_some()) + && !flush_active + { let database = self.env.d1("DB")?; begin_d1_flush(&mut attachment).map(|snapshot| (database, snapshot)) } else { @@ -912,7 +924,7 @@ impl DurableObject for LiveHub { ws.serialize_attachment(&attachment)?; } - if !attachment.hidden { + if !fresh_reports.is_empty() && !attachment.hidden { let mut sockets = self.state.get_websockets_with_tag("all"); sockets.extend( self.state @@ -935,11 +947,13 @@ impl DurableObject for LiveHub { .history .point() .ok_or_else(|| Error::RustError("missing live history aggregate".to_string()))?; + let sample_count = snapshot.history.sample_count(); let metrics_result = crate::db::save_reports_with_history( &database, &server_id, - &batch.samples, + &fresh_reports, &history_point, + sample_count, &traffic_state, &snapshot.latency, ) diff --git a/src/routes/agent.rs b/src/routes/agent.rs index 28ced2d..fd5f4f8 100644 --- a/src/routes/agent.rs +++ b/src/routes/agent.rs @@ -85,10 +85,7 @@ pub(crate) async fn report( if latency_result_count > MAX_AGENT_LATENCY_RESULTS { return error("每批最多包含 4096 条延迟结果", 400); } - let mut history_aggregate = db::HistoryMetricAggregate::default(); - history_aggregate.extend(&batch.samples); - let alert_point = history_aggregate.point(); - db::save_reports(database, &server_id, &batch.samples).await?; + let alert_point = db::save_reports(database, &server_id, &batch.samples).await?; if let Some(point) = alert_point.as_ref() { if let Err(error) = live::record_alert_sample(&env, &server_id, point).await { worker::console_warn!("resource alert sample write failed: {error:?}"); diff --git a/wrangler.jsonc b/wrangler.jsonc index e1739c8..84369b4 100644 --- a/wrangler.jsonc +++ b/wrangler.jsonc @@ -2,7 +2,7 @@ "$schema": "./node_modules/wrangler/config-schema.json", "name": "nodeflare", "main": "build/index.js", - "compatibility_date": "2026-08-17", + "compatibility_date": "2026-08-18", "keep_vars": true, "workers_dev": true, "preview_urls": false,