feat: add resource monitoring and provider upgrade

2026-05-04 13:04:39 +00:00
parent 8726611b6f
commit 2a03f6343c
27 changed files with 1807 additions and 130 deletions
@@ -5,6 +5,8 @@ import postgres from "postgres";
 import {
  type ApiEvent,
  type ApiNode,
+  type ApiNodeDockerStatus,
+  type ApiNodeSystemStatus,
  type ApiTask,
  type CoreDispatchMessage,
  type JsonValue,
@@ -146,6 +148,36 @@ async function initDatabase(client: SqlClient): Promise<void> {
      updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
    )
  `;
+  await client`
+    CREATE TABLE IF NOT EXISTS unidesk_node_docker_status (
+      provider_id TEXT PRIMARY KEY,
+      status JSONB NOT NULL DEFAULT '{}'::jsonb,
+      collected_at TIMESTAMPTZ,
+      updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+    )
+  `;
+  await client`
+    CREATE TABLE IF NOT EXISTS unidesk_node_system_status (
+      provider_id TEXT PRIMARY KEY,
+      status JSONB NOT NULL DEFAULT '{}'::jsonb,
+      collected_at TIMESTAMPTZ,
+      updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+    )
+  `;
+  await client`
+    CREATE TABLE IF NOT EXISTS unidesk_node_metric_samples (
+      id BIGSERIAL PRIMARY KEY,
+      provider_id TEXT NOT NULL,
+      collected_at TIMESTAMPTZ NOT NULL,
+      cpu_percent DOUBLE PRECISION NOT NULL DEFAULT 0,
+      memory_percent DOUBLE PRECISION NOT NULL DEFAULT 0,
+      disk_percent DOUBLE PRECISION NOT NULL DEFAULT 0,
+      sample JSONB NOT NULL DEFAULT '{}'::jsonb,
+      created_at TIMESTAMPTZ NOT NULL DEFAULT now()
+    )
+  `;
+  await client`CREATE INDEX IF NOT EXISTS idx_unidesk_node_system_status_updated_at ON unidesk_node_system_status(updated_at DESC)`;
+  await client`CREATE INDEX IF NOT EXISTS idx_unidesk_node_metric_samples_provider_time ON unidesk_node_metric_samples(provider_id, collected_at DESC)`;
  dbReady = true;
  logger("info", "database_init_complete");
 }
@@ -242,6 +274,57 @@ async function touchHeartbeat(providerId: string, labels: ProviderLabels): Promi
  `;
 }

+async function upsertDockerStatus(providerId: string, status: JsonValue, collectedAt: string): Promise<void> {
+  await sql`
+    INSERT INTO unidesk_node_docker_status (provider_id, status, collected_at, updated_at)
+    VALUES (${providerId}, ${sql.json(status)}, ${collectedAt}, now())
+    ON CONFLICT (provider_id) DO UPDATE SET
+      status = EXCLUDED.status,
+      collected_at = EXCLUDED.collected_at,
+      updated_at = now()
+  `;
+}
+
+function recordValue(value: unknown, key: string): unknown {
+  if (typeof value !== "object" || value === null || Array.isArray(value)) return undefined;
+  return (value as Record<string, unknown>)[key];
+}
+
+function nestedNumber(value: unknown, objectKey: string, numberKey: string): number {
+  const parsed = Number(recordValue(recordValue(value, objectKey), numberKey));
+  return Number.isFinite(parsed) ? parsed : 0;
+}
+
+async function upsertSystemStatus(providerId: string, status: JsonValue, collectedAt: string): Promise<void> {
+  const cpuPercent = nestedNumber(status, "cpu", "percent");
+  const memoryPercent = nestedNumber(status, "memory", "percent");
+  const diskPercent = nestedNumber(status, "disk", "percent");
+  await sql.begin(async (tx) => {
+    await tx`
+      INSERT INTO unidesk_node_system_status (provider_id, status, collected_at, updated_at)
+      VALUES (${providerId}, ${tx.json(status)}, ${collectedAt}, now())
+      ON CONFLICT (provider_id) DO UPDATE SET
+        status = EXCLUDED.status,
+        collected_at = EXCLUDED.collected_at,
+        updated_at = now()
+    `;
+    await tx`
+      INSERT INTO unidesk_node_metric_samples (provider_id, collected_at, cpu_percent, memory_percent, disk_percent, sample)
+      VALUES (${providerId}, ${collectedAt}, ${cpuPercent}, ${memoryPercent}, ${diskPercent}, ${tx.json(status)})
+    `;
+    await tx`
+      DELETE FROM unidesk_node_metric_samples
+      WHERE provider_id = ${providerId}
+        AND id NOT IN (
+          SELECT id FROM unidesk_node_metric_samples
+          WHERE provider_id = ${providerId}
+          ORDER BY collected_at DESC
+          LIMIT 720
+        )
+    `;
+  });
+}
+
 async function markProviderOffline(providerId: string): Promise<void> {
  activeProviders.delete(providerId);
  if (!dbReady) return;
@@ -302,9 +385,42 @@ async function handleProviderMessage(ws: ProviderSocket, raw: string | Buffer):
    return;
  }

+  if (message.type === "system_status") {
+    await upsertSystemStatus(message.providerId, message.status as unknown as JsonValue, message.status.collectedAt);
+    logger("debug", "provider_system_status", {
+      providerId: message.providerId,
+      cpuPercent: message.status.cpu.percent,
+      memoryPercent: message.status.memory.percent,
+      diskPercent: message.status.disk.percent,
+      ok: message.status.ok,
+    });
+    return;
+  }
+
+  if (message.type === "docker_status") {
+    await upsertDockerStatus(message.providerId, message.status as unknown as JsonValue, message.status.collectedAt);
+    logger("debug", "provider_docker_status", { providerId: message.providerId, counts: message.status.counts, ok: message.status.ok });
+    return;
+  }
+
  await sql`
+    WITH incoming AS (
+      SELECT ${message.status}::text AS status, ${sql.json(message.result ?? { message: message.message })}::jsonb AS result
+    )
    UPDATE unidesk_tasks
-    SET status = ${message.status}, result = ${sql.json(message.result ?? { message: message.message })}, updated_at = now()
+    SET
+      status = CASE
+        WHEN unidesk_tasks.status IN ('succeeded', 'failed') AND incoming.status NOT IN ('succeeded', 'failed') THEN unidesk_tasks.status
+        WHEN unidesk_tasks.status = 'running' AND incoming.status = 'accepted' THEN unidesk_tasks.status
+        ELSE incoming.status
+      END,
+      result = CASE
+        WHEN unidesk_tasks.status IN ('succeeded', 'failed') AND incoming.status NOT IN ('succeeded', 'failed') THEN unidesk_tasks.result
+        WHEN unidesk_tasks.status = 'running' AND incoming.status = 'accepted' THEN unidesk_tasks.result
+        ELSE incoming.result
+      END,
+      updated_at = now()
+    FROM incoming
    WHERE id = ${message.taskId}
  `;
  await recordEvent("task_status", message.providerId, {
@@ -332,6 +448,74 @@ async function getNodes(): Promise<ApiNode[]> {
  }));
 }

+async function getNodeDockerStatuses(): Promise<ApiNodeDockerStatus[]> {
+  const rows = await sql<Array<Record<string, unknown>>>`
+    SELECT n.provider_id, n.name, n.status AS node_status, d.status AS docker_status, d.updated_at
+    FROM unidesk_nodes n
+    LEFT JOIN unidesk_node_docker_status d ON d.provider_id = n.provider_id
+    ORDER BY n.status DESC, n.provider_id ASC
+  `;
+  return rows.map((row) => ({
+    providerId: String(row.provider_id),
+    name: String(row.name),
+    nodeStatus: row.node_status === "online" ? "online" : "offline",
+    dockerStatus: row.docker_status === null || row.docker_status === undefined ? null : (row.docker_status as JsonValue),
+    updatedAt: row.updated_at instanceof Date ? row.updated_at.toISOString() : row.updated_at === null || row.updated_at === undefined ? null : String(row.updated_at),
+  }));
+}
+
+function metricPointFromSample(sample: unknown, collectedAt: string): JsonValue {
+  return {
+    at: collectedAt,
+    cpuPercent: nestedNumber(sample, "cpu", "percent"),
+    memoryPercent: nestedNumber(sample, "memory", "percent"),
+    diskPercent: nestedNumber(sample, "disk", "percent"),
+    memoryUsedBytes: nestedNumber(sample, "memory", "usedBytes"),
+    memoryTotalBytes: nestedNumber(sample, "memory", "totalBytes"),
+    diskUsedBytes: nestedNumber(sample, "disk", "usedBytes"),
+    diskTotalBytes: nestedNumber(sample, "disk", "totalBytes"),
+    load1: nestedNumber(sample, "cpu", "load1"),
+  };
+}
+
+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
+    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
+  `;
+  const sampleRows = await sql<Array<Record<string, unknown>>>`
+    SELECT provider_id, collected_at, sample
+    FROM (
+      SELECT provider_id, collected_at, sample,
+        row_number() OVER (PARTITION BY provider_id ORDER BY collected_at DESC) AS rn
+      FROM unidesk_node_metric_samples
+    ) ranked
+    WHERE rn <= ${limit}
+    ORDER BY provider_id ASC, collected_at ASC
+  `;
+  const historyByProvider = new Map<string, JsonValue[]>();
+  for (const row of sampleRows) {
+    const providerId = String(row.provider_id);
+    const collectedAt = row.collected_at instanceof Date ? row.collected_at.toISOString() : String(row.collected_at);
+    const history = historyByProvider.get(providerId) ?? [];
+    history.push(metricPointFromSample(row.sample ?? {}, collectedAt));
+    historyByProvider.set(providerId, history);
+  }
+  return currentRows.map((row) => {
+    const providerId = String(row.provider_id);
+    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),
+      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),
+    };
+  });
+}
+
 async function getEvents(limit: number): Promise<ApiEvent[]> {
  const rows = await sql<Array<Record<string, unknown>>>`
    SELECT id, type, source, payload, created_at
@@ -370,6 +554,8 @@ async function getTasks(limit: number): Promise<ApiTask[]> {
 async function getOverview(): Promise<JsonValue> {
  const nodes = await getNodes();
  const tasks = await getTasks(50);
+  const dockerStatuses = await getNodeDockerStatuses();
+  const systemStatuses = await getNodeSystemStatuses(1);
  const online = nodes.filter((node) => node.status === "online").length;
  const pendingTasks = tasks.filter((task) => task.status === "queued" || task.status === "dispatched" || task.status === "running").length;
  return {
@@ -379,6 +565,8 @@ async function getOverview(): Promise<JsonValue> {
    uptimeSeconds: Math.floor((Date.now() - serviceStartedAt.getTime()) / 1000),
    nodeCount: nodes.length,
    onlineNodeCount: online,
+    dockerStatusNodeCount: dockerStatuses.filter((item) => item.dockerStatus !== null).length,
+    systemStatusNodeCount: systemStatuses.filter((item) => item.current !== null).length,
    pendingTaskCount: pendingTasks,
    activeSocketCount: activeProviders.size,
    heartbeatTimeoutMs: config.heartbeatTimeoutMs,
@@ -388,7 +576,7 @@ async function getOverview(): Promise<JsonValue> {
 async function dispatchTask(req: Request): Promise<Response> {
  const body = (await req.json()) as { providerId?: unknown; command?: unknown; payload?: unknown };
  const providerId = typeof body.providerId === "string" ? body.providerId : "";
-  const command = body.command === "docker.ps" || body.command === "echo" ? body.command : "echo";
+  const command = body.command === "docker.ps" || body.command === "provider.upgrade" || body.command === "echo" ? body.command : "echo";
  const payload = typeof body.payload === "object" && body.payload !== null ? (body.payload as Record<string, JsonValue>) : {};
  if (!providerId) {
    return jsonResponse({ ok: false, error: "providerId is required" }, 400);
@@ -405,7 +593,11 @@ async function dispatchTask(req: Request): Promise<Response> {
  }
  const dispatch: CoreDispatchMessage = { type: "dispatch", taskId, command, payload };
  socket.send(JSON.stringify(dispatch));
-  await sql`UPDATE unidesk_tasks SET status = 'dispatched', updated_at = now() WHERE id = ${taskId}`;
+  await sql`
+    UPDATE unidesk_tasks
+    SET status = 'dispatched', updated_at = now()
+    WHERE id = ${taskId} AND status = 'queued'
+  `;
  await recordEvent("task_dispatched", providerId, { taskId, providerId, command });
  return jsonResponse({ ok: true, taskId, status: "dispatched", providerOnline: true });
 }
@@ -423,6 +615,8 @@ async function route(req: Request): Promise<Response> {
    }
    if (url.pathname === "/api/overview") return jsonResponse(await getOverview());
    if (url.pathname === "/api/nodes") return jsonResponse({ ok: true, nodes: await getNodes() });
+    if (url.pathname === "/api/nodes/system-status") return jsonResponse({ ok: true, systemStatuses: await getNodeSystemStatuses(readLimit(url, 120)) });
+    if (url.pathname === "/api/nodes/docker-status") return jsonResponse({ ok: true, dockerStatuses: await getNodeDockerStatuses() });
    if (url.pathname === "/api/events") return jsonResponse({ ok: true, events: await getEvents(readLimit(url, 100)) });
    if (url.pathname === "/api/tasks") return jsonResponse({ ok: true, tasks: await getTasks(readLimit(url, 100)) });
    if (url.pathname === "/api/dispatch" && req.method === "POST") return dispatchTask(req);