From 7ca313a9086ac3395391a17c15704335a8b4190e Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 13 May 2026 11:47:08 +0000 Subject: [PATCH] Improve microservice health availability --- src/components/backend-core/src/index.ts | 314 ++++++++++++++++++++++- src/components/frontend/src/app.tsx | 13 +- src/components/frontend/src/claudeqq.tsx | 12 +- 3 files changed, 336 insertions(+), 3 deletions(-) diff --git a/src/components/backend-core/src/index.ts b/src/components/backend-core/src/index.ts index b0634c23..70585591 100644 --- a/src/components/backend-core/src/index.ts +++ b/src/components/backend-core/src/index.ts @@ -185,6 +185,18 @@ interface MicroserviceProxyCacheEntry { bodyText: string; } +interface MicroserviceHealthAssessment { + healthy: boolean; + reason: string; + checks: Record; + body: JsonValue; +} + +interface MicroserviceAvailabilityEntry { + expiresAt: number; + probe: Record; +} + const recentLogs: unknown[] = []; const activeProviders = new Map(); const activeSshClients = new Map(); @@ -204,10 +216,13 @@ const maxPerformanceSamples = 3000; const taskTerminalWaiters = new Map>(); const microserviceProxyCache = new Map(); const microserviceProxyRefreshes = new Map>(); +const microserviceAvailabilityCache = new Map(); +const microserviceAvailabilityRefreshes = new Map>>(); const activeScheduledRuns = new Set(); let lastTaskSweepAt = 0; let taskSweepInFlight: Promise | null = null; const microserviceProxyMaxBodyTextLength = 8 * 1024 * 1024; +const microserviceAvailabilityTtlMs = 30_000; const microserviceForwardRequestHeaders = [ "accept", "content-type", @@ -1445,7 +1460,7 @@ async function getPgdataUsage(): Promise { } async function getOverview(): Promise { - const [nodeRows, dockerRows, systemRows, pendingTasks, pgdata] = await Promise.all([ + const [nodeRows, dockerRows, systemRows, pendingTasks, pgdata, microserviceAvailabilitySummary] = await Promise.all([ sql>` SELECT count(*)::int AS node_count, @@ -1464,6 +1479,7 @@ async function getOverview(): Promise { `, countPendingTasks(), getPgdataUsage(), + getMicroserviceAvailabilitySummary(), ]); const nodeRow = nodeRows[0]; const dockerRow = dockerRows[0]; @@ -1479,6 +1495,7 @@ async function getOverview(): Promise { dockerStatusNodeCount: Number(dockerRow?.docker_status_node_count ?? 0), systemStatusNodeCount: Number(systemRow?.system_status_node_count ?? 0), pendingTaskCount: pendingTasks, + microserviceAvailability: microserviceAvailabilitySummary, taskPendingTimeoutMs: config.taskPendingTimeoutMs, activeSocketCount: activeProviders.size, heartbeatTimeoutMs: config.heartbeatTimeoutMs, @@ -1503,6 +1520,38 @@ function findContainer(status: JsonValue | null, containerName: string): JsonVal return container === undefined ? null : container as JsonValue; } +function inferredMicroserviceAvailability(service: MicroserviceConfig, providerStatus: string, container: JsonValue | null): JsonValue { + const cached = cachedMicroserviceAvailability(service.id); + if (cached !== null) return cached; + const containerState = isPlainRecord(container) ? String(container.state ?? "").toLowerCase() : ""; + const containerStatus = isPlainRecord(container) ? String(container.status ?? "") : ""; + if (providerStatus !== "online") { + return { + serviceId: service.id, + healthy: false, + status: "unhealthy", + reason: `provider is ${providerStatus}`, + source: "runtime-inference", + }; + } + if (container !== null && containerState !== "running") { + return { + serviceId: service.id, + healthy: false, + status: "unhealthy", + reason: `container is ${containerStatus || containerState || "not running"}`, + source: "runtime-inference", + }; + } + return { + serviceId: service.id, + healthy: null, + status: "unknown", + reason: "strict health has not been probed yet", + source: "runtime-inference", + }; +} + async function getMicroservices(): Promise { const nodes = await getNodes(); const dockerStatuses = await getNodeDockerStatuses(); @@ -1517,6 +1566,7 @@ async function getMicroservices(): Promise { providerName: node?.name ?? service.providerId, providerLastHeartbeat: node?.lastHeartbeat ?? null, container, + availability: inferredMicroserviceAvailability(service, node?.status ?? "missing", container), backendPortMapping: { providerId: service.providerId, node: `${service.backend.nodeBindHost}:${service.backend.nodePort}`, @@ -2464,7 +2514,267 @@ function responseFromMicroserviceResult(task: Awaited }); } +function isPlainRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function nestedRecord(value: unknown, key: string): Record | null { + if (!isPlainRecord(value)) return null; + const child = value[key]; + return isPlainRecord(child) ? child : null; +} + +function nestedBoolean(value: unknown, key: string): boolean | null { + if (!isPlainRecord(value)) return null; + const child = value[key]; + return typeof child === "boolean" ? child : null; +} + +function nestedString(value: unknown, key: string): string | null { + if (!isPlainRecord(value)) return null; + const child = value[key]; + return typeof child === "string" ? child : null; +} + +function jsonBodyFromText(bodyText: string, contentType: string): JsonValue { + if (!contentTypeIsJson(contentType)) return bodyText.length > 0 ? truncateText(bodyText, 4000) : null; + try { + return compactJson(JSON.parse(bodyText) as unknown); + } catch { + return bodyText.length > 0 ? truncateText(bodyText, 4000) : null; + } +} + +function healthFailure(body: JsonValue, checks: Record, reason: string): MicroserviceHealthAssessment { + return { healthy: false, reason, checks, body }; +} + +function healthSuccess(body: JsonValue, checks: Record, reason = "health endpoint returned a usable service state"): MicroserviceHealthAssessment { + return { healthy: true, reason, checks, body }; +} + +function genericMicroserviceHealthAssessment(upstreamStatus: number, body: JsonValue): MicroserviceHealthAssessment { + const checks: Record = { + upstreamStatus, + upstream2xx: upstreamStatus >= 200 && upstreamStatus < 300, + }; + if (upstreamStatus < 200 || upstreamStatus >= 300) { + return healthFailure(body, checks, `health endpoint returned HTTP ${upstreamStatus}`); + } + if (isPlainRecord(body)) { + const ok = body.ok; + const success = body.success; + const statusText = typeof body.status === "string" ? body.status : ""; + checks.okField = typeof ok === "boolean" ? ok : null; + checks.successField = typeof success === "boolean" ? success : null; + checks.statusField = statusText || null; + if (ok === false) return healthFailure(body, checks, "health body ok=false"); + if (success === false) return healthFailure(body, checks, "health body success=false"); + if (["error", "failed", "fail", "unhealthy", "down", "offline", "not_ready", "not-ready"].includes(statusText.toLowerCase())) { + return healthFailure(body, checks, `health body status=${statusText}`); + } + } + return healthSuccess(body, checks); +} + +function assessMetNonlinearHealth(upstreamStatus: number, body: JsonValue): MicroserviceHealthAssessment { + const base = genericMicroserviceHealthAssessment(upstreamStatus, body); + const checks = { ...base.checks }; + if (!base.healthy) return { ...base, checks }; + if (!isPlainRecord(body)) return healthFailure(body, checks, "MET Nonlinear health body is not JSON"); + checks.service = typeof body.service === "string" ? body.service : null; + checks.queueReady = isPlainRecord(body.queue); + const image = nestedRecord(body, "image"); + checks.mlImage = typeof image?.image === "string" ? image.image : null; + checks.mlImagePresent = image?.present === true; + if (body.service !== "met-nonlinear-unidesk-ts") { + return healthFailure(body, checks, "MET Nonlinear health returned an unexpected service identity"); + } + if (!isPlainRecord(body.queue)) return healthFailure(body, checks, "MET Nonlinear queue state is not available"); + if (image?.present !== true) return healthFailure(body, checks, "MET Nonlinear ML image is not ready"); + return healthSuccess(body, checks, "MET Nonlinear backend, queue, and ML image are ready"); +} + +function assessClaudeQqHealth(upstreamStatus: number, body: JsonValue): MicroserviceHealthAssessment { + const base = genericMicroserviceHealthAssessment(upstreamStatus, body); + const checks = { ...base.checks }; + if (!base.healthy) return { ...base, checks }; + if (!isPlainRecord(body)) return healthFailure(body, checks, "ClaudeQQ health body is not JSON"); + const napcat = nestedRecord(body, "napcat"); + const endpoints = Array.isArray(body.endpoints) ? body.endpoints.map((item) => String(item)) : []; + const httpConnected = nestedBoolean(napcat, "httpConnected") === true; + const wsConnected = nestedBoolean(napcat, "wsConnected") === true; + const napcatConnected = nestedBoolean(napcat, "connected") === true; + const loginState = nestedString(napcat, "loginState") ?? ""; + checks.service = typeof body.service === "string" ? body.service : null; + checks.pushEndpoint = endpoints.includes("/api/push/text"); + checks.napcatHttpLoggedIn = httpConnected; + checks.napcatWebsocketConnected = wsConnected; + checks.napcatConnected = napcatConnected; + checks.loginState = loginState || null; + if (body.service !== "claudeqq") return healthFailure(body, checks, "ClaudeQQ health returned an unexpected service identity"); + if (!endpoints.includes("/api/push/text")) return healthFailure(body, checks, "ClaudeQQ push API endpoint is not advertised"); + if (!httpConnected) return healthFailure(body, checks, "NapCat HTTP is not logged in"); + if (!wsConnected) return healthFailure(body, checks, "NapCat OneBot WebSocket is not connected"); + if (!napcatConnected || loginState !== "logged_in") return healthFailure(body, checks, "NapCat is not in a logged-in send/receive ready state"); + return healthSuccess(body, checks, "ClaudeQQ and NapCat are logged in and send/receive ready"); +} + +function assessMicroserviceHealth(service: MicroserviceConfig, upstreamStatus: number, contentType: string, bodyText: string): MicroserviceHealthAssessment { + const body = jsonBodyFromText(bodyText, contentType); + if (service.id === "met-nonlinear") return assessMetNonlinearHealth(upstreamStatus, body); + if (service.id === "claudeqq") return assessClaudeQqHealth(upstreamStatus, body); + return genericMicroserviceHealthAssessment(upstreamStatus, body); +} + +function healthProbeFromAssessment( + service: MicroserviceConfig, + upstreamStatus: number, + contentType: string, + assessment: MicroserviceHealthAssessment, +): Record { + return { + serviceId: service.id, + name: service.name, + providerId: service.providerId, + healthPath: service.backend.healthPath, + healthy: assessment.healthy, + status: assessment.healthy ? "healthy" : "unhealthy", + reason: assessment.reason, + upstreamStatus, + contentType, + checks: assessment.checks, + checkedAt: new Date().toISOString(), + }; +} + +function rememberMicroserviceAvailability(serviceId: string, probe: Record): void { + microserviceAvailabilityCache.set(serviceId, { expiresAt: Date.now() + microserviceAvailabilityTtlMs, probe }); +} + +function cachedMicroserviceAvailability(serviceId: string): Record | null { + const entry = microserviceAvailabilityCache.get(serviceId); + if (entry === undefined) return null; + if (entry.expiresAt <= Date.now()) { + microserviceAvailabilityCache.delete(serviceId); + return null; + } + return entry.probe; +} + +async function strictMicroserviceHealthProbe(service: MicroserviceConfig, timeoutMs?: number): Promise> { + const probeService = timeoutMs === undefined + ? service + : { ...service, backend: { ...service.backend, timeoutMs } }; + const response = await fetchMicroserviceUpstreamResponse(probeService, "GET", service.backend.healthPath, { query: "", jsonArrayLimits: {} }, { accept: "application/json" }, ""); + const upstreamStatus = response.status; + const contentType = response.headers.get("content-type") ?? "application/octet-stream"; + const bodyText = await response.text(); + const assessment = assessMicroserviceHealth(service, upstreamStatus, contentType, bodyText); + const probe = healthProbeFromAssessment(service, upstreamStatus, contentType, assessment); + rememberMicroserviceAvailability(service.id, probe); + return { ...probe, body: assessment.body }; +} + +async function strictMicroserviceHealthResponse(service: MicroserviceConfig, headOnly: boolean): Promise { + const response = await fetchMicroserviceUpstreamResponse(service, "GET", service.backend.healthPath, { query: "", jsonArrayLimits: {} }, { accept: "application/json" }, ""); + const upstreamStatus = response.status; + const contentType = response.headers.get("content-type") ?? "application/octet-stream"; + const bodyText = await response.text(); + const assessment = assessMicroserviceHealth(service, upstreamStatus, contentType, bodyText); + const probe = healthProbeFromAssessment(service, upstreamStatus, contentType, assessment); + rememberMicroserviceAvailability(service.id, probe); + if (assessment.healthy) { + return new Response(headOnly ? null : bodyText, { + status: upstreamStatus, + headers: { + "content-type": contentType, + "x-unidesk-health": "healthy", + "x-unidesk-health-reason": assessment.reason, + }, + }); + } + const status = upstreamStatus >= 500 ? upstreamStatus : 503; + return jsonResponse({ + ok: false, + status: "unhealthy", + serviceId: service.id, + name: service.name, + providerId: service.providerId, + reason: assessment.reason, + checkedAt: probe.checkedAt, + checks: assessment.checks, + upstream: { + status: upstreamStatus, + contentType, + body: assessment.body, + }, + }, status); +} + +async function microserviceAvailability(service: MicroserviceConfig): Promise> { + const cached = cachedMicroserviceAvailability(service.id); + if (cached !== null) return cached; + const existing = microserviceAvailabilityRefreshes.get(service.id); + if (existing !== undefined) return existing; + const timeoutMs = Math.max(2500, Math.min(service.backend.timeoutMs, 10_000)); + const refresh = strictMicroserviceHealthProbe(service, timeoutMs) + .catch((error) => { + const probe = { + serviceId: service.id, + name: service.name, + providerId: service.providerId, + healthPath: service.backend.healthPath, + healthy: false, + status: "unhealthy", + reason: error instanceof Error ? error.message : String(error), + upstreamStatus: null, + contentType: null, + checks: {}, + checkedAt: new Date().toISOString(), + } satisfies Record; + rememberMicroserviceAvailability(service.id, probe); + return probe; + }) + .finally(() => { + microserviceAvailabilityRefreshes.delete(service.id); + }); + microserviceAvailabilityRefreshes.set(service.id, refresh); + return refresh; +} + +async function getMicroserviceAvailabilitySummary(): Promise> { + const probes = await Promise.all(config.microservices.map((service) => microserviceAvailability(service))); + const healthy = probes.filter((probe) => probe.healthy === true); + const byProvider: Record = {}; + for (const probe of probes) { + const providerId = String(probe.providerId ?? "unknown"); + const current = isPlainRecord(byProvider[providerId]) ? byProvider[providerId] as Record : { total: 0, healthy: 0, unhealthy: 0 }; + current.total = Number(current.total ?? 0) + 1; + if (probe.healthy === true) current.healthy = Number(current.healthy ?? 0) + 1; + else current.unhealthy = Number(current.unhealthy ?? 0) + 1; + byProvider[providerId] = current; + } + return { + totalCount: probes.length, + healthyCount: healthy.length, + unhealthyCount: probes.length - healthy.length, + checkedAt: new Date().toISOString(), + byProvider, + services: probes.map((probe) => ({ + serviceId: probe.serviceId, + name: probe.name, + providerId: probe.providerId, + healthy: probe.healthy, + status: probe.status, + reason: probe.reason, + checkedAt: probe.checkedAt, + })), + }; +} + function microserviceCacheTtlMs(serviceId: string, targetPath: string): number { + if (targetPath === "/health") return 0; if (serviceId === "pipeline" && targetPath === "/api/snapshot") return 6_000; if (serviceId === "pipeline" && targetPath.startsWith("/api/oa-event-flow/")) return 20_000; if (serviceId === "pipeline" && targetPath.startsWith("/api/model-quota/")) return 60_000; @@ -2487,6 +2797,7 @@ function microserviceCacheStaleMs(serviceId: string, targetPath: string): number } function providerMicroserviceCacheTtlMs(serviceId: string, targetPath: string): number { + if (targetPath === "/health") return 0; if (serviceId === "met-nonlinear" && (targetPath === "/api/images" || targetPath === "/api/projects")) return 60_000; if (serviceId === "met-nonlinear" && targetPath === "/api/history") return 10_000; if (serviceId === "met-nonlinear" && (targetPath === "/api/queue" || targetPath === "/api/summary")) return 3_000; @@ -2695,6 +3006,7 @@ async function microserviceRoute(req: Request, url: URL): Promise { if (!isMicroservicePathAllowed(service, targetPath)) { return jsonResponse({ ok: false, error: "microservice path is not allowed", serviceId, targetPath }, 403); } + if (suffix === "health") return strictMicroserviceHealthResponse(service, method === "HEAD"); const proxyOptions = readMicroserviceArrayLimits(url); const cacheKey = microserviceCacheKey(service, method, targetPath, proxyOptions); const cacheTtlMs = microserviceCacheTtlMs(service.id, targetPath); diff --git a/src/components/frontend/src/app.tsx b/src/components/frontend/src/app.tsx index ac4f49da..4802ac5c 100644 --- a/src/components/frontend/src/app.tsx +++ b/src/components/frontend/src/app.tsx @@ -619,6 +619,10 @@ function OverviewPage({ data, onRaw, onNavigate }: AnyRecord) { const pendingCount = overview.pendingTaskCount ?? pendingTasks.length; const recentTasks = data.tasks.slice(0, 5); const pgdata = overview.pgdata || {}; + const microserviceAvailability = overview.microserviceAvailability || {}; + const microserviceTotal = asNumber(microserviceAvailability.totalCount); + const microserviceHealthy = asNumber(microserviceAvailability.healthyCount); + const microserviceUnhealthy = asNumber(microserviceAvailability.unhealthyCount); return h("div", { className: "page-grid overview-grid", "data-testid": "overview-page" }, h(Panel, { title: "核心指标", eyebrow: "Control" }, h("div", { className: "metric-grid" }, @@ -626,6 +630,7 @@ function OverviewPage({ data, onRaw, onNavigate }: AnyRecord) { h(MetricCard, { label: "PGDATA", value: fmtBytes(pgdata.databaseBytes), hint: `${pgdata.volumeName || "unidesk_pgdata_10gb"} / ${pgdata.databasePretty || "--"}`, tone: "ok", testId: "pgdata-usage-card" }), h(MetricCard, { label: "在线节点", value: overview.onlineNodeCount ?? 0, hint: `${overview.nodeCount ?? 0} registered`, tone: "ok" }), h(MetricCard, { label: "WebSocket", value: overview.activeSocketCount ?? 0, hint: "Provider ingress sockets" }), + h(MetricCard, { label: "用户服务可用", value: microserviceTotal > 0 ? `${microserviceHealthy}/${microserviceTotal}` : "--", hint: microserviceTotal > 0 ? `healthyCount ${microserviceHealthy} · unhealthyCount ${microserviceUnhealthy}` : "strict /health probes", tone: microserviceTotal > 0 && microserviceUnhealthy === 0 ? "ok" : "warn", testId: "microservice-availability-card" }), h(MetricCard, { label: "待处理任务", value: pendingCount, hint: pendingCount > 0 ? "点击查看具体任务" : `timeout ${fmtDuration(Math.floor((overview.taskPendingTimeoutMs ?? 0) / 1000))}`, tone: pendingCount > 0 ? "warn" : "ok", onClick: () => onNavigate("tasks", "pending"), testId: "pending-task-card" }), ), ), @@ -1608,6 +1613,8 @@ function MicroserviceCatalogPage({ microservices, onRaw, onNavigate }: AnyRecord const runtime = microserviceRuntime(service); const repository = microserviceRepository(service); const backend = microserviceBackend(service); + const availability = runtime.availability || {}; + const availabilityStatus = availability.status || (runtime.providerStatus === "online" ? "unknown" : "unhealthy"); return h("tr", { key: service.id, "data-testid": `microservice-row-${safeId(service.id)}` }, h("td", null, h("strong", null, service.name), h("code", null, service.id)), h("td", null, h("strong", null, runtime.providerName || service.providerId), h("code", null, service.providerId)), @@ -1618,7 +1625,11 @@ function MicroserviceCatalogPage({ microservices, onRaw, onNavigate }: AnyRecord h("code", null, `${backend.nodeBindHost || "--"}:${backend.nodePort || "--"} -> ${backend.proxyMode || "--"}`), ), h("td", null, h("span", null, service.development?.sshPassthrough ? "SSH 透传" : "未配置"), h("code", null, service.development?.worktreePath || "--")), - h("td", null, h(StatusBadge, { status: runtime.providerStatus === "online" ? "online" : "warn" }, runtime.providerStatus || "unknown"), h(DataSummary, { data: runtime.container, empty: "容器快照未上报" })), + h("td", null, + h(StatusBadge, { status: availabilityStatus === "healthy" ? "online" : availabilityStatus === "unknown" ? "warn" : "failed" }, availabilityStatus), + h("span", null, availability.reason || runtime.providerStatus || "unknown"), + h(DataSummary, { data: runtime.container, empty: "容器快照未上报" }), + ), h("td", null, h("div", { className: "microservice-actions" }, service.id === "findjob" ? h("button", { type: "button", className: "ghost-btn", onClick: () => onNavigate("apps", "findjob"), "data-testid": "open-findjob-button" }, "打开") : null, diff --git a/src/components/frontend/src/claudeqq.tsx b/src/components/frontend/src/claudeqq.tsx index 282ac5c1..8acc1eb7 100644 --- a/src/components/frontend/src/claudeqq.tsx +++ b/src/components/frontend/src/claudeqq.tsx @@ -22,6 +22,16 @@ async function requestJson(path: string, options: AnyRecord = {}): Promise return requestUniDeskJson(path, { failureFields: ["ok", "success"], ...options }); } +async function requestHealthJson(path: string): Promise { + const response = await fetch(path, { credentials: "same-origin" }); + const text = await response.text(); + try { + return text ? JSON.parse(text) : { ok: response.ok, status: response.status }; + } catch { + return { ok: response.ok, status: response.status, text }; + } +} + function StatusBadge({ status, children }: AnyRecord) { const normalized = String(status || "unknown").toLowerCase(); return h("span", { className: `status-badge ${normalized}` }, children || status || "unknown"); @@ -119,7 +129,7 @@ export function ClaudeQqPage({ microservices, onRaw, apiBaseUrl = "/api" }: AnyR setState((prev: any) => ({ ...prev, loading: true, error: "" })); try { const [health, status, events, subscriptions, sent] = await Promise.all([ - requestJson(`${apiBaseUrl}/microservices/claudeqq/health`), + requestHealthJson(`${apiBaseUrl}/microservices/claudeqq/health`), requestJson(claudeqqApi(apiBaseUrl, "/api/server/status")), requestJson(claudeqqApi(apiBaseUrl, "/api/events/recent?limit=60")), requestJson(claudeqqApi(apiBaseUrl, "/api/events/subscriptions")),