mirrors/pikasTech-unidesk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: restore node resource status sync

Browse Source
This commit is contained in:
Codex
2026-06-12 14:18:55 +00:00
parent 17b54c685a
commit b41847853a
11 changed files with 347 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/components/backend-core/src/db.rs
+62 -8
View File
@@ -1,7 +1,7 @@
use std::sync::Arc;
use anyhow::Context;
use chrono::{DateTime, Utc};
use chrono::{DateTime, Duration, Utc};
use serde_json::{json, Value};
use tokio_postgres::Row;
@@ -192,18 +192,43 @@ pub async fn update_provider_heartbeat(
Ok(())
}
fn provider_collected_at(
state: &Arc<AppState>,
provider_id: &str,
source: &str,
value: &str,
) -> DateTime<Utc> {
match DateTime::parse_from_rfc3339(value) {
Ok(parsed) => parsed.with_timezone(&Utc),
Err(error) => {
state.log(
"warn",
"provider_status_collected_at_invalid",
Some(json!({
"providerId": provider_id,
"source": source,
"value": value,
"error": error.to_string()
})),
);
Utc::now()
}
}
}
pub async fn upsert_docker_status(
state: &Arc<AppState>,
provider_id: &str,
status: &JsonValue,
collected_at: &str,
) -> anyhow::Result<()> {
let collected_at = provider_collected_at(state, provider_id, "docker_status", collected_at);
let client = state.pool.get().await?;
client
.execute(
r#"
INSERT INTO unidesk_node_docker_status (provider_id, status, collected_at, updated_at)
VALUES ($1, $2, $3::timestamptz, now())
VALUES ($1, $2, $3, now())
ON CONFLICT (provider_id) DO UPDATE SET
status = EXCLUDED.status,
collected_at = EXCLUDED.collected_at,
@@ -224,12 +249,13 @@ pub async fn upsert_system_status(
let cpu_percent = nested_number(status, "cpu", "percent");
let memory_percent = nested_number(status, "memory", "percent");
let disk_percent = nested_number(status, "disk", "percent");
let collected_at = provider_collected_at(state, provider_id, "system_status", collected_at);
let mut client = state.pool.get().await?;
let tx = client.transaction().await?;
tx.execute(
r#"
INSERT INTO unidesk_node_system_status (provider_id, status, collected_at, updated_at)
VALUES ($1, $2, $3::timestamptz, now())
VALUES ($1, $2, $3, now())
ON CONFLICT (provider_id) DO UPDATE SET
status = EXCLUDED.status,
collected_at = EXCLUDED.collected_at,
@@ -241,10 +267,18 @@ pub async fn upsert_system_status(
tx.execute(
r#"
INSERT INTO unidesk_node_metric_samples (provider_id, collected_at, cpu_percent, memory_percent, disk_percent, sample)
VALUES ($1, $2::timestamptz, $3, $4, $5, $6)
VALUES ($1, $2, $3, $4, $5, $6)
"#,
&[&provider_id, &collected_at, &cpu_percent, &memory_percent, &disk_percent, status],
).await?;
&[
&provider_id,
&collected_at,
&cpu_percent,
&memory_percent,
&disk_percent,
status,
],
)
.await?;
tx.execute(
r#"
DELETE FROM unidesk_node_metric_samples
@@ -321,6 +355,18 @@ fn metric_point_from_sample(sample: &JsonValue, collected_at: &str) -> JsonValue
})
}
const SYSTEM_STATUS_STALE_AFTER_SECONDS: i64 = 300;
fn system_status_is_stale(collected_at: Option<DateTime<Utc>>, now: DateTime<Utc>) -> bool {
collected_at
.map(|value| now.signed_duration_since(value) > Duration::seconds(SYSTEM_STATUS_STALE_AFTER_SECONDS))
.unwrap_or(false)
}
fn system_status_age_seconds(collected_at: Option<DateTime<Utc>>, now: DateTime<Utc>) -> Option<i64> {
collected_at.map(|value| now.signed_duration_since(value).num_seconds().max(0))
}
pub async fn get_node_system_statuses(
state: &Arc<AppState>,
limit: i64,
@@ -328,7 +374,7 @@ pub async fn get_node_system_statuses(
let client = state.pool.get().await?;
let current_rows = client.query(
r#"
SELECT n.provider_id, n.name, n.status AS node_status, s.status AS system_status, s.updated_at
SELECT n.provider_id, n.name, n.status AS node_status, s.status AS system_status, s.collected_at AS system_collected_at, s.updated_at
FROM unidesk_nodes n
LEFT JOIN unidesk_node_system_status s ON s.provider_id = n.provider_id
ORDER BY n.status DESC, n.provider_id ASC
@@ -366,15 +412,23 @@ pub async fn get_node_system_statuses(
&collected_at.to_rfc3339_opts(chrono::SecondsFormat::Millis, true),
));
}
let now = Utc::now();
Ok(current_rows.into_iter().map(|row| {
let provider_id: String = row.get("provider_id");
let updated_at: Option<DateTime<Utc>> = row.get("updated_at");
let system_collected_at: Option<DateTime<Utc>> = row.get("system_collected_at");
let system_status: Option<JsonValue> = row.get("system_status");
let stale = system_status_is_stale(system_collected_at, now);
let current = if stale { None } else { system_status.clone() };
json!({
"providerId": provider_id,
"name": row.get::<_, String>("name"),
"nodeStatus": if row.get::<_, String>("node_status") == "online" { "online" } else { "offline" },
"current": system_status,
"current": current,
"lastKnown": system_status,
"currentCollectedAt": system_collected_at.map(|value| value.to_rfc3339_opts(chrono::SecondsFormat::Millis, true)),
"stale": stale,
"staleSeconds": system_status_age_seconds(system_collected_at, now),
"history": history_by_provider.remove(&provider_id).unwrap_or_default(),
"updatedAt": updated_at.map(|value| value.to_rfc3339_opts(chrono::SecondsFormat::Millis, true)),
})
src/components/backend-core/src/db.ts
+23 -2
View File
@@ -237,9 +237,22 @@ function metricPointFromSample(sample: unknown, collectedAt: string): JsonValue
};
}
const SYSTEM_STATUS_STALE_AFTER_SECONDS = 300;
function systemStatusAgeSeconds(collectedAt: unknown, now: Date): number | null {
const date = collectedAt instanceof Date ? collectedAt : typeof collectedAt === "string" ? new Date(collectedAt) : null;
if (date === null || Number.isNaN(date.getTime())) return null;
return Math.max(0, Math.floor((now.getTime() - date.getTime()) / 1000));
}
function systemStatusIsStale(collectedAt: unknown, now: Date): boolean {
const ageSeconds = systemStatusAgeSeconds(collectedAt, now);
return ageSeconds !== null && ageSeconds > SYSTEM_STATUS_STALE_AFTER_SECONDS;
}
export async function getNodeSystemStatuses(limit: number): Promise<ApiNodeSystemStatus[]> {
const currentRows = await sql()<Array<Record<string, unknown>>>`
SELECT n.provider_id, n.name, n.status AS node_status, s.status AS system_status, s.updated_at
SELECT n.provider_id, n.name, n.status AS node_status, s.status AS system_status, s.collected_at AS system_collected_at, s.updated_at
FROM unidesk_nodes n
LEFT JOIN unidesk_node_system_status s ON s.provider_id = n.provider_id
ORDER BY n.status DESC, n.provider_id ASC
@@ -265,13 +278,21 @@ export async function getNodeSystemStatuses(limit: number): Promise<ApiNodeSyste
history.push(metricPointFromSample(row.sample ?? {}, collectedAt));
historyByProvider.set(providerId, history);
}
const now = new Date();
return currentRows.map((row) => {
const providerId = String(row.provider_id);
const lastKnown = row.system_status === null || row.system_status === undefined ? null : (row.system_status as JsonValue);
const currentCollectedAt = row.system_collected_at instanceof Date ? row.system_collected_at.toISOString() : row.system_collected_at === null || row.system_collected_at === undefined ? null : String(row.system_collected_at);
const stale = systemStatusIsStale(row.system_collected_at, now);
return {
providerId,
name: String(row.name),
nodeStatus: row.node_status === "online" ? "online" : "offline",
current: row.system_status === null || row.system_status === undefined ? null : (row.system_status as JsonValue),
current: stale ? null : lastKnown,
lastKnown,
currentCollectedAt,
stale,
staleSeconds: systemStatusAgeSeconds(row.system_collected_at, now),
history: historyByProvider.get(providerId) ?? [],
updatedAt: row.updated_at instanceof Date ? row.updated_at.toISOString() : row.updated_at === null || row.updated_at === undefined ? null : String(row.updated_at),
};
src/components/backend-core/src/provider_registry.rs
+33 -8
View File
@@ -571,22 +571,22 @@ async fn provider_socket_task(state: Arc<AppState>, socket: WebSocket) {
while let Some(message) = receiver_ws.next().await {
match message {
Ok(Message::Text(text)) => {
if let Err(error) = handle_provider_text(&state, &connection, text).await {
if let Err(error) = handle_provider_text(&state, &connection, &text).await {
state.log(
"error",
"provider_message_failed",
Some(json!({ "error": error.to_string() })),
Some(provider_message_error_context(&text, &error)),
);
let _ = tx.send(Message::Text(json!({ "type": "ack", "requestId": "message", "ok": false, "message": error.to_string() }).to_string()));
}
}
Ok(Message::Binary(bytes)) => {
if let Ok(text) = String::from_utf8(bytes.to_vec()) {
if let Err(error) = handle_provider_text(&state, &connection, text).await {
if let Err(error) = handle_provider_text(&state, &connection, &text).await {
state.log(
"error",
"provider_message_failed",
Some(json!({ "error": error.to_string() })),
Some(provider_message_error_context(&text, &error)),
);
}
}
@@ -629,12 +629,36 @@ async fn provider_socket_task(state: Arc<AppState>, socket: WebSocket) {
send_task.abort();
}
fn provider_message_error_context(text: &str, error: &anyhow::Error) -> Value {
let parsed = serde_json::from_str::<Value>(text).ok();
json!({
"error": error.to_string(),
"messageType": parsed
.as_ref()
.and_then(|value| value.get("type"))
.and_then(Value::as_str)
.unwrap_or("unknown"),
"providerId": parsed
.as_ref()
.and_then(|value| value.get("providerId"))
.and_then(Value::as_str)
.unwrap_or("unknown"),
"requestId": parsed
.as_ref()
.and_then(|value| value.get("requestId"))
.and_then(Value::as_str)
.unwrap_or("unknown"),
"textBytes": text.len(),
"textPreview": text.chars().take(300).collect::<String>(),
})
}
async fn handle_provider_text(
state: &Arc<AppState>,
connection: &Arc<ProviderConnection>,
text: String,
text: &str,
) -> anyhow::Result<()> {
let message: Value = serde_json::from_str(&text)?;
let message: Value = serde_json::from_str(text)?;
let provider_id = message
.get("providerId")
.and_then(Value::as_str)
@@ -845,6 +869,7 @@ pub async fn mark_stale_providers_offline(state: &Arc<AppState>) -> anyhow::Resu
return Ok(());
}
let timeout_ms = state.config.heartbeat_timeout_ms as i64;
let cutoff = Utc::now() - Duration::milliseconds(timeout_ms);
let client = state.pool.get().await?;
let rows = client
.query(
@@ -853,10 +878,10 @@ pub async fn mark_stale_providers_offline(state: &Arc<AppState>) -> anyhow::Resu
SET status = 'offline', updated_at = now()
WHERE status = 'online'
AND last_heartbeat IS NOT NULL
AND last_heartbeat < now() - ($1 * interval '1 millisecond')
AND last_heartbeat < $1
RETURNING provider_id
"#,
&[&timeout_ms],
&[&cutoff],
)
.await?;
for row in rows {
src/components/frontend/public/app.js
+75 -75
View File
File diff suppressed because one or more lines are too long
src/components/frontend/src/app.tsx
+48 -11
View File
@@ -784,11 +784,50 @@ function HeartbeatPage({ nodes }: AnyRecord) {
);
}
function metricPointFromCurrent(current: AnyRecord): AnyRecord {
const cpu = current?.cpu || {};
const memory = current?.memory || {};
const disk = current?.disk || {};
return {
at: current?.collectedAt,
cpuPercent: asNumber(cpu.percent),
memoryPercent: asNumber(memory.percent),
diskPercent: asNumber(disk.percent),
memoryUsedBytes: asNumber(memory.usedBytes),
memoryTotalBytes: asNumber(memory.totalBytes),
diskUsedBytes: asNumber(disk.usedBytes),
diskTotalBytes: asNumber(disk.totalBytes),
load1: asNumber(cpu.load1),
};
}
function synchronizedMetricPoints(history: any[], current: AnyRecord | null): AnyRecord[] {
if (!current) return [];
const currentPoint = metricPointFromCurrent(current);
const currentAt = timeMs(currentPoint.at);
const points = (Array.isArray(history) ? history : [])
.filter((point) => point && typeof point === "object")
.filter((point) => currentAt === null || timeMs(point.at) !== currentAt);
points.push(currentPoint);
return points
.sort((left, right) => (timeMs(left.at) ?? 0) - (timeMs(right.at) ?? 0))
.slice(-60);
}
function NodeMonitorPage({ nodes, systemStatuses, tasks, onRaw, refresh }: AnyRecord) {
const [selectedProvider, setSelectedProvider] = useState("");
const merged = useMemo(() => nodes.map((node: any) => {
const status = systemStatuses.find((item: any) => item.providerId === node.providerId);
return { ...node, systemCurrent: status?.current || null, systemHistory: status?.history || [], systemUpdatedAt: status?.updatedAt || null };
return {
...node,
systemCurrent: status?.current || null,
systemLastKnown: status?.lastKnown || null,
systemCurrentCollectedAt: status?.currentCollectedAt || null,
systemStale: Boolean(status?.stale),
systemStaleSeconds: status?.staleSeconds ?? null,
systemHistory: status?.history || [],
systemUpdatedAt: status?.updatedAt || null,
};
}), [nodes, systemStatuses]);
const active = merged.find((node: any) => node.providerId === selectedProvider) || merged[0] || null;
useEffect(() => {
@@ -802,12 +841,10 @@ function NodeMonitorPage({ nodes, systemStatuses, tasks, onRaw, refresh }: AnyRe
const cpu = current?.cpu || {};
const memory = current?.memory || {};
const disk = current?.disk || {};
const points = history.length > 0 ? history : current ? [{
at: current.collectedAt,
cpuPercent: asNumber(cpu.percent),
memoryPercent: asNumber(memory.percent),
diskPercent: asNumber(disk.percent),
}] : [];
const points = synchronizedMetricPoints(history, current);
const staleText = active.systemCurrentCollectedAt
? `最后采样 ${fmtDate(active.systemCurrentCollectedAt)}，已过期 ${fmtDuration(asNumber(active.systemStaleSeconds))}`
: "最后采样时间不可用";
return h("div", { className: "monitor-page", "data-testid": "node-monitor-page" },
h("div", { className: "docker-node-strip" },
@@ -820,7 +857,7 @@ function NodeMonitorPage({ nodes, systemStatuses, tasks, onRaw, refresh }: AnyRe
h("span", { className: `pulse ${node.status}` }),
h("strong", null, node.name),
h("code", null, node.providerId),
h("span", null, node.systemCurrent ? `CPU ${fmtPercent(node.systemCurrent.cpu?.percent)} / MEM ${fmtPercent(node.systemCurrent.memory?.percent)}` : "等待指标"),
h("span", null, node.systemCurrent ? `CPU ${fmtPercent(node.systemCurrent.cpu?.percent)} / MEM ${fmtPercent(node.systemCurrent.memory?.percent)}` : node.systemStale ? "指标过期" : "等待指标"),
)),
),
h("div", { className: "monitor-layout" },
@@ -828,9 +865,9 @@ function NodeMonitorPage({ nodes, systemStatuses, tasks, onRaw, refresh }: AnyRe
title: "任务管理器视图",
eyebrow: active.name,
className: "monitor-main-panel",
actions: current ? h(RawButton, { title: `System ${active.providerId}`, data: { current, history }, onOpen: onRaw }) : null,
actions: current || active.systemStale ? h(RawButton, { title: `System ${active.providerId}`, data: { current, lastKnown: active.systemLastKnown, currentCollectedAt: active.systemCurrentCollectedAt, stale: active.systemStale, staleSeconds: active.systemStaleSeconds, history }, onOpen: onRaw }) : null,
},
!current ? h(EmptyState, { title: "系统指标未上报", text: "provider-gateway 会周期性采集 /proc 与 df，并保存历史曲线" }) :
!current ? h(EmptyState, { title: active.systemStale ? "系统指标已过期" : "系统指标未上报", text: active.systemStale ? `${staleText}；等待 provider-gateway 恢复 system.status 上报` : "provider-gateway 会周期性采集 /proc 与 df，并保存历史曲线" }) :
h("div", null,
h("div", { className: "monitor-hero" },
h("div", null,
@@ -854,7 +891,7 @@ function NodeMonitorPage({ nodes, systemStatuses, tasks, onRaw, refresh }: AnyRe
h(MetricCard, { label: "CPU 当前", value: fmtPercent(cpu.percent), hint: `history ${points.length} samples`, tone: "ok" }),
h(MetricCard, { label: "实际内存", value: fmtBytes(memory.usedBytes), hint: `${fmtPercent(memory.percent)} 不含缓存` }),
h(MetricCard, { label: "硬盘已用", value: fmtBytes(disk.usedBytes), hint: fmtPercent(disk.percent) }),
h(MetricCard, { label: "更新时间", value: fmtDate(active.systemUpdatedAt || current.collectedAt), hint: active.providerId }),
h(MetricCard, { label: "更新时间", value: fmtDate(current.collectedAt || active.systemUpdatedAt), hint: active.providerId }),
),
h(ProcessResourceTable, { current, onRaw }),
),
src/components/shared/src/index.ts
+4
View File
@@ -278,6 +278,10 @@ export interface ApiNodeSystemStatus {
name: string;
nodeStatus: "online" | "offline";
current: JsonValue | null;
lastKnown?: JsonValue | null;
currentCollectedAt?: string | null;
stale?: boolean;
staleSeconds?: number | null;
history: JsonValue[];
updatedAt: string | null;
}