Improve microservice health availability

2026-05-13 11:47:08 +00:00
parent 1f811f3c07
commit 7ca313a908
3 changed files with 336 additions and 3 deletions
@@ -185,6 +185,18 @@ interface MicroserviceProxyCacheEntry {
  bodyText: string;
 }

+interface MicroserviceHealthAssessment {
+  healthy: boolean;
+  reason: string;
+  checks: Record<string, JsonValue>;
+  body: JsonValue;
+}
+
+interface MicroserviceAvailabilityEntry {
+  expiresAt: number;
+  probe: Record<string, JsonValue>;
+}
+
 const recentLogs: unknown[] = [];
 const activeProviders = new Map<string, ProviderSocket>();
 const activeSshClients = new Map<string, ProviderSocket>();
@@ -204,10 +216,13 @@ const maxPerformanceSamples = 3000;
 const taskTerminalWaiters = new Map<string, Set<TaskTerminalWaiter>>();
 const microserviceProxyCache = new Map<string, MicroserviceProxyCacheEntry>();
 const microserviceProxyRefreshes = new Map<string, Promise<void>>();
+const microserviceAvailabilityCache = new Map<string, MicroserviceAvailabilityEntry>();
+const microserviceAvailabilityRefreshes = new Map<string, Promise<Record<string, JsonValue>>>();
 const activeScheduledRuns = new Set<string>();
 let lastTaskSweepAt = 0;
 let taskSweepInFlight: Promise<void> | null = null;
 const microserviceProxyMaxBodyTextLength = 8 * 1024 * 1024;
+const microserviceAvailabilityTtlMs = 30_000;
 const microserviceForwardRequestHeaders = [
  "accept",
  "content-type",
@@ -1445,7 +1460,7 @@ async function getPgdataUsage(): Promise<JsonValue> {
 }

 async function getOverview(): Promise<JsonValue> {
-  const [nodeRows, dockerRows, systemRows, pendingTasks, pgdata] = await Promise.all([
+  const [nodeRows, dockerRows, systemRows, pendingTasks, pgdata, microserviceAvailabilitySummary] = await Promise.all([
    sql<Array<{ node_count: string | number; online_node_count: string | number }>>`
      SELECT
        count(*)::int AS node_count,
@@ -1464,6 +1479,7 @@ async function getOverview(): Promise<JsonValue> {
    `,
    countPendingTasks(),
    getPgdataUsage(),
+    getMicroserviceAvailabilitySummary(),
  ]);
  const nodeRow = nodeRows[0];
  const dockerRow = dockerRows[0];
@@ -1479,6 +1495,7 @@ async function getOverview(): Promise<JsonValue> {
    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<JsonValue[]> {
  const nodes = await getNodes();
  const dockerStatuses = await getNodeDockerStatuses();
@@ -1517,6 +1566,7 @@ async function getMicroservices(): Promise<JsonValue[]> {
        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<ReturnType<typeof rawTask>
  });
 }

+function isPlainRecord(value: unknown): value is Record<string, unknown> {
+  return typeof value === "object" && value !== null && !Array.isArray(value);
+}
+
+function nestedRecord(value: unknown, key: string): Record<string, unknown> | 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<string, JsonValue>, reason: string): MicroserviceHealthAssessment {
+  return { healthy: false, reason, checks, body };
+}
+
+function healthSuccess(body: JsonValue, checks: Record<string, JsonValue>, 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<string, JsonValue> = {
+    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<string, JsonValue> {
+  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<string, JsonValue>): void {
+  microserviceAvailabilityCache.set(serviceId, { expiresAt: Date.now() + microserviceAvailabilityTtlMs, probe });
+}
+
+function cachedMicroserviceAvailability(serviceId: string): Record<string, JsonValue> | 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<Record<string, JsonValue>> {
+  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<Response> {
+  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<Record<string, JsonValue>> {
+  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<string, JsonValue>;
+      rememberMicroserviceAvailability(service.id, probe);
+      return probe;
+    })
+    .finally(() => {
+      microserviceAvailabilityRefreshes.delete(service.id);
+    });
+  microserviceAvailabilityRefreshes.set(service.id, refresh);
+  return refresh;
+}
+
+async function getMicroserviceAvailabilitySummary(): Promise<Record<string, JsonValue>> {
+  const probes = await Promise.all(config.microservices.map((service) => microserviceAvailability(service)));
+  const healthy = probes.filter((probe) => probe.healthy === true);
+  const byProvider: Record<string, JsonValue> = {};
+  for (const probe of probes) {
+    const providerId = String(probe.providerId ?? "unknown");
+    const current = isPlainRecord(byProvider[providerId]) ? byProvider[providerId] as Record<string, JsonValue> : { 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<Response> {
  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);