Codex
342 lines
17 KiB
TypeScript
342 lines
17 KiB
TypeScript
// SPEC: PJ2026-01060307 控制面模块化 draft-2026-06-25-p0. runtime-common module for scripts/src/hwlab-node-impl.ts.
|
|
|
|
// Moved mechanically from scripts/src/hwlab-node-impl.ts:1600-1885 for #903.
|
|
|
|
// SPEC: PJ2026-01060505 Workbench Performance draft-2026-06-17-p0.
|
|
// SPEC: PJ2026-01060508 Web哨兵 draft-2026-06-25-p0-web-probe-sentinel.
|
|
// Responsibility: YAML-first node/lane operations, including Workbench observability control commands.
|
|
import { createHash, randomBytes } from "node:crypto";
|
|
import { existsSync, mkdirSync, readFileSync, renameSync, writeFileSync } from "node:fs";
|
|
import { dirname, join } from "node:path";
|
|
import { repoRoot, rootPath, type Config } from "../config";
|
|
import { runCommand, type CommandResult } from "../command";
|
|
import { startJob } from "../jobs";
|
|
import { classifySshTcpPoolFailure } from "../ssh";
|
|
import { HWLAB_NODE_CONTROL_PLANE_CONFIG_PATH, hwlabNodeControlPlaneInfraHelp, runHwlabNodeControlPlaneInfra } from "../hwlab-node-control-plane";
|
|
import { hwlabRuntimeLaneConfigPath, hwlabRuntimeLaneIds, hwlabRuntimeLaneSpec, hwlabRuntimeLaneSpecForNode, hwlabRuntimeNodeIds, isHwlabRuntimeLane, type HwlabRuntimeLane, type HwlabRuntimeLaneSpec, type HwlabRuntimeObservabilityRecordingRuleSpec, type HwlabRuntimeObservabilitySpec, type HwlabRuntimeObservabilityWarningAlertSpec, type HwlabRuntimePublicExposureSpec, type HwlabRuntimeWebProbeAlertThresholdsSpec, type HwlabRuntimeWebProbeProjectManagementSpec } from "../hwlab-node-lanes";
|
|
import { nodeWebProbeScriptRunnerSource } from "../hwlab-node-web-probe-runner-source";
|
|
import { nodeWebObserveAnalyzerSource } from "../hwlab-node-web-observe-analyzer-source";
|
|
import { nodeWebObserveRunnerSource } from "../hwlab-node-web-observe-runner-source";
|
|
import { nodeWebObserveCollectViewNodeScript, parseNodeWebProbeObserveCollectView, type NodeWebProbeObserveCollectView } from "../hwlab-node-web-observe-collect";
|
|
import { withWebObserveCollectRendered, withWebObserveCommandRendered, withWebObserveStatusRendered } from "../hwlab-node-web-observe-render";
|
|
import { buildWebObserveWrapperForObserveOptions, webObserveWrapperStateDirFromStatus } from "../hwlab-node-web-observe-wrapper";
|
|
import { renderWebObserveWrapperContract } from "../hwlab-node-web-observe-wrapper-render";
|
|
import { runWebProbeSentinelCommand, type WebProbeSentinelOptions } from "../hwlab-node-web-sentinel-cicd";
|
|
import { hwlabNodeHelp, hwlabNodeObservabilityHelp, hwlabNodeWebProbeHelp } from "../hwlab-node-help";
|
|
import { compactWebProbeResult, compactWebProbeScriptResult } from "../hwlab-node-web-probe-summary";
|
|
import { nodeObservabilityRecordingRuleExpression, nodeObservabilityRecordingRuleSummaries, nodeObservabilityWarningAlertExpression, nodeObservabilityWarningAlertSummaries } from "../hwlab-node-observability-promql";
|
|
import { runDelegatedHwlabNodeCommand, type DelegatedNodeDomain } from "../hwlab-node-transport";
|
|
import type { RenderedCliResult } from "../output";
|
|
|
|
import type { NodeRuntimeGitMirrorTargetSpec } from "./entry";
|
|
import { commandResultFromAsync, isCommandSuccess } from "./cleanup";
|
|
import { nodeRuntimeExpected, parseNodeScopedDelegatedOptions } from "./plan";
|
|
import { nodeRuntimeRenderToken } from "./render";
|
|
import { parseJsonObject, record, shellQuote } from "./utils";
|
|
|
|
export function metricNameFromPrometheusLine(line: string): string | null {
|
|
const match = /^([A-Za-z_:][A-Za-z0-9_:]*)/u.exec(line);
|
|
return match?.[1] ?? null;
|
|
}
|
|
|
|
export function prometheusLineMatchesMetric(line: string, name: string): boolean {
|
|
if (!line.startsWith(name)) return false;
|
|
const next = line.charAt(name.length);
|
|
return next === "" || next === "{" || /\s/u.test(next);
|
|
}
|
|
|
|
export function prometheusLineHasLabel(line: string, name: string, value: string): boolean {
|
|
return line.includes(`${name}="${value.replace(/\\/gu, "\\\\").replace(/"/gu, '\\"')}"`);
|
|
}
|
|
|
|
export function isWorkbenchBackendEventMetric(name: string): boolean {
|
|
return name.startsWith("hwlab_workbench_event_") || name.startsWith("hwlab_workbench_backend_event_");
|
|
}
|
|
|
|
export function compactPrometheusLines(lines: readonly string[]): string[] {
|
|
return lines.slice(0, 16).map((line) => line.length > 320 ? `${line.slice(0, 317)}...` : line);
|
|
}
|
|
|
|
export function parseJsonRecordFromText(text: string): Record<string, unknown> {
|
|
const trimmed = text.trim();
|
|
if (trimmed.length === 0) return {};
|
|
try {
|
|
const parsed = JSON.parse(trimmed) as unknown;
|
|
return record(parsed);
|
|
} catch {
|
|
const start = trimmed.indexOf("{");
|
|
const end = trimmed.lastIndexOf("}");
|
|
if (start >= 0 && end > start) {
|
|
try {
|
|
return record(JSON.parse(trimmed.slice(start, end + 1)) as unknown);
|
|
} catch {}
|
|
}
|
|
}
|
|
return {};
|
|
}
|
|
|
|
export function k8sServiceSelector(serviceJson: Record<string, unknown>): Record<string, string> {
|
|
const spec = record(serviceJson.spec);
|
|
const selector = record(spec.selector);
|
|
return Object.fromEntries(Object.entries(selector).filter((entry): entry is [string, string] => typeof entry[1] === "string" && entry[1].length > 0));
|
|
}
|
|
|
|
export function labelSelectorText(selector: Record<string, string>): string {
|
|
return Object.entries(selector).map(([key, value]) => `${key}=${value}`).join(",");
|
|
}
|
|
|
|
export function selectK8sPod(podsJson: Record<string, unknown>, containerName: string): { name: string; phase: string | null; ready: boolean; containerReady: boolean } | null {
|
|
const items = Array.isArray(podsJson.items) ? podsJson.items.map(record) : [];
|
|
const pods = items.map((item) => {
|
|
const metadata = record(item.metadata);
|
|
const status = record(item.status);
|
|
const name = typeof metadata.name === "string" ? metadata.name : "";
|
|
const phase = typeof status.phase === "string" ? status.phase : null;
|
|
const containerStatuses = Array.isArray(status.containerStatuses) ? status.containerStatuses.map(record) : [];
|
|
const targetContainer = containerStatuses.find((container) => container.name === containerName);
|
|
const containerReady = targetContainer?.ready === true;
|
|
return {
|
|
name,
|
|
phase,
|
|
ready: name.length > 0 && phase === "Running" && containerReady,
|
|
containerReady,
|
|
};
|
|
}).filter((pod) => pod.name.length > 0);
|
|
return pods.find((pod) => pod.ready) ?? pods.find((pod) => pod.phase === "Running") ?? pods[0] ?? null;
|
|
}
|
|
|
|
export function compactRuntimeCommand(result: CommandResult): Record<string, unknown> {
|
|
return {
|
|
exitCode: result.exitCode,
|
|
stdoutBytes: Buffer.byteLength(result.stdout),
|
|
stderrBytes: Buffer.byteLength(result.stderr),
|
|
stdoutTail: result.stdout.trim().slice(-2000),
|
|
stderrTail: result.stderr.trim().slice(-2000),
|
|
timedOut: result.timedOut,
|
|
};
|
|
}
|
|
|
|
export function compactRuntimeCommandStats(result: CommandResult): Record<string, unknown> {
|
|
return {
|
|
exitCode: result.exitCode,
|
|
stdoutBytes: Buffer.byteLength(result.stdout),
|
|
stderrBytes: Buffer.byteLength(result.stderr),
|
|
timedOut: result.timedOut,
|
|
};
|
|
}
|
|
|
|
export function parseLastJsonLineObject(text: string): Record<string, unknown> {
|
|
for (const line of text.split(/\r?\n/u).reverse()) {
|
|
const trimmed = line.trim();
|
|
if (!trimmed.startsWith("{") || !trimmed.endsWith("}")) continue;
|
|
try {
|
|
return record(JSON.parse(trimmed) as unknown);
|
|
} catch {
|
|
// Keep scanning: command tails can mix kubectl status, git output, and JSON payloads.
|
|
}
|
|
}
|
|
return {};
|
|
}
|
|
|
|
export function escapeRegExp(value: string): string {
|
|
return value.replace(/[.*+?^${}()|[\]\\]/gu, "\\$&");
|
|
}
|
|
|
|
export function nodeRuntimeGitMirrorRefSources(scoped: ReturnType<typeof parseNodeScopedDelegatedOptions>, mirror: NodeRuntimeGitMirrorTargetSpec): Record<string, unknown> {
|
|
return {
|
|
localSource: `refs/heads/${mirror.sourceBranch}`,
|
|
githubSource: `refs/mirror-stage/heads/${mirror.sourceBranch}`,
|
|
localGitops: `refs/heads/${mirror.gitopsBranch}`,
|
|
githubGitops: `refs/mirror-stage/heads/${mirror.gitopsBranch}`,
|
|
githubFieldsAreMirrorStageCache: true,
|
|
cacheRefresh: `bun scripts/cli.ts hwlab nodes git-mirror sync --node ${scoped.node} --lane ${scoped.lane} --confirm --wait`,
|
|
};
|
|
}
|
|
|
|
export function nodeRuntimeGitMirrorFlushPartialSuccess(scoped: ReturnType<typeof parseNodeScopedDelegatedOptions>, mirror: NodeRuntimeGitMirrorTargetSpec, result: CommandResult): Record<string, unknown> | null {
|
|
if (scoped.action !== "flush") return null;
|
|
const text = `${result.stdout}\n${result.stderr}`;
|
|
const payload = parseLastJsonLineObject(text);
|
|
const partialSuccess = typeof payload.partialSuccess === "string" && payload.partialSuccess.length > 0
|
|
? payload.partialSuccess
|
|
: null;
|
|
const payloadStatus = typeof payload.status === "string" && payload.status.length > 0 ? payload.status : null;
|
|
const branch = escapeRegExp(mirror.gitopsBranch);
|
|
const pushSucceededInGitOutput = new RegExp(`\\b${branch}\\s*->\\s*${branch}\\b`, "u").test(text);
|
|
const postPushFetchFailed = /kex_exchange_identification|Connection closed by remote host|Could not read from remote repository|fatal:.*fetch|fetch-pack|early EOF/iu.test(text);
|
|
if (partialSuccess !== "push-succeeded-fetch-failed" && !(result.exitCode !== 0 && pushSucceededInGitOutput && postPushFetchFailed)) {
|
|
return null;
|
|
}
|
|
return {
|
|
status: payloadStatus ?? "partial-success",
|
|
partialSuccess: "push-succeeded-fetch-failed",
|
|
degradedReason: "node-runtime-git-mirror-flush-post-push-fetch-failed",
|
|
retryable: true,
|
|
message: "GitOps push appears to have succeeded, but the post-push fetch/recheck failed. Refresh mirror-stage cache before deciding another flush is needed.",
|
|
payload: Object.keys(payload).length > 0 ? payload : null,
|
|
refSources: nodeRuntimeGitMirrorRefSources(scoped, mirror),
|
|
next: {
|
|
sync: `bun scripts/cli.ts hwlab nodes git-mirror sync --node ${scoped.node} --lane ${scoped.lane} --confirm --wait`,
|
|
status: `bun scripts/cli.ts hwlab nodes git-mirror status --node ${scoped.node} --lane ${scoped.lane}`,
|
|
},
|
|
};
|
|
}
|
|
|
|
export function sshTcpPoolDiagnosticsFromCommand(spec: HwlabRuntimeLaneSpec, result: CommandResult): Record<string, unknown> | null {
|
|
if (isCommandSuccess(result)) return null;
|
|
const failureKind = classifySshTcpPoolFailure(`${result.stderr}\n${result.stdout}`);
|
|
if (failureKind === null) return null;
|
|
return {
|
|
classification: "ssh-tcp-pool-transient",
|
|
failureKind,
|
|
providerId: spec.nodeId,
|
|
route: spec.nodeKubeRoute,
|
|
message: "SSH tcp data pool failed while running the controlled command; treat this as transport/data-pool transient until provider labels and retry say otherwise.",
|
|
next: {
|
|
poolStatus: `bun scripts/cli.ts debug ssh-pool ${spec.nodeId}`,
|
|
retrySmoke: `trans ${spec.nodeId} argv true`,
|
|
retryApply: `bun scripts/cli.ts hwlab nodes control-plane apply --node ${spec.nodeId} --lane ${spec.lane} --confirm`,
|
|
retryTriggerCurrent: `bun scripts/cli.ts hwlab nodes control-plane trigger-current --node ${spec.nodeId} --lane ${spec.lane} --confirm`,
|
|
},
|
|
};
|
|
}
|
|
|
|
export function nodeRuntimeUnsupportedAction(scoped: ReturnType<typeof parseNodeScopedDelegatedOptions>): Record<string, unknown> {
|
|
return {
|
|
ok: false,
|
|
command: `hwlab nodes control-plane ${scoped.action} --node ${scoped.node} --lane ${scoped.lane}`,
|
|
node: scoped.node,
|
|
lane: scoped.lane,
|
|
mutation: false,
|
|
degradedReason: "unsupported-node-scoped-runtime-action",
|
|
message: "node-scoped runtime currently supports plan/status/apply/refresh/sync/trigger-current/cleanup-runs/runtime-image/runtime-migration/source-workspace",
|
|
expected: nodeRuntimeExpected(scoped.spec),
|
|
};
|
|
}
|
|
|
|
export function transPath(): string {
|
|
return join(repoRoot, "scripts", "trans");
|
|
}
|
|
|
|
export function runNodeHostScript(spec: HwlabRuntimeLaneSpec, script: string, timeoutSeconds: number, input = ""): CommandResult {
|
|
return runCommand([transPath(), spec.nodeRoute, "sh", "--", script], repoRoot, { input, timeoutMs: timeoutSeconds * 1000 });
|
|
}
|
|
|
|
export function runNodeHostScriptAsync(spec: HwlabRuntimeLaneSpec, script: string, timeoutSeconds: number, label: string): CommandResult {
|
|
const token = nodeRuntimeRenderToken();
|
|
const stateDir = `/tmp/unidesk-hwlab-node-runtime/${label}-${token}`;
|
|
const statusPath = `${stateDir}/status.json`;
|
|
const stdoutPath = `${stateDir}/stdout.log`;
|
|
const stderrPath = `${stateDir}/stderr.log`;
|
|
const runnerPath = `${stateDir}/run.sh`;
|
|
const startScript = [
|
|
"set -eu",
|
|
`state_dir=${shellQuote(stateDir)}`,
|
|
`status_path=${shellQuote(statusPath)}`,
|
|
`stdout_path=${shellQuote(stdoutPath)}`,
|
|
`stderr_path=${shellQuote(stderrPath)}`,
|
|
`runner_path=${shellQuote(runnerPath)}`,
|
|
"rm -rf \"$state_dir\"",
|
|
"mkdir -p \"$state_dir\"",
|
|
"cat > \"$runner_path\" <<'UNIDESK_REMOTE_RUNNER'",
|
|
script,
|
|
"UNIDESK_REMOTE_RUNNER",
|
|
"chmod 700 \"$runner_path\"",
|
|
"python3 - \"$status_path\" <<'PY'",
|
|
"import datetime, json, sys",
|
|
"json.dump({'state':'running','ok':None,'pid':None,'startedAt':datetime.datetime.utcnow().isoformat()+'Z','finishedAt':None,'exitCode':None}, open(sys.argv[1], 'w'), indent=2)",
|
|
"PY",
|
|
"(",
|
|
" set +e",
|
|
" sh \"$runner_path\" >\"$stdout_path\" 2>\"$stderr_path\"",
|
|
" code=$?",
|
|
" python3 - \"$status_path\" \"$stdout_path\" \"$stderr_path\" \"$code\" <<'PY'",
|
|
"import datetime, json, pathlib, sys",
|
|
"status_path, stdout_path, stderr_path, code = sys.argv[1:5]",
|
|
"def tail(path):",
|
|
" try:",
|
|
" text = pathlib.Path(path).read_text(errors='replace')",
|
|
" except FileNotFoundError:",
|
|
" return ''",
|
|
" return text[-12000:]",
|
|
"payload = {'state':'finished','ok':code == '0','pid':None,'startedAt':None,'finishedAt':datetime.datetime.utcnow().isoformat()+'Z','exitCode':int(code),'stdout':tail(stdout_path),'stderr':tail(stderr_path)}",
|
|
"try:",
|
|
" current = json.load(open(status_path))",
|
|
" payload['startedAt'] = current.get('startedAt')",
|
|
"except Exception:",
|
|
" pass",
|
|
"json.dump(payload, open(status_path, 'w'), indent=2)",
|
|
"PY",
|
|
") >/dev/null 2>&1 &",
|
|
"pid=$!",
|
|
"python3 - \"$status_path\" \"$pid\" <<'PY'",
|
|
"import json, sys",
|
|
"path, pid = sys.argv[1:3]",
|
|
"payload = json.load(open(path))",
|
|
"payload['pid'] = int(pid)",
|
|
"json.dump(payload, open(path, 'w'), indent=2)",
|
|
"PY",
|
|
"printf '{\"ok\":true,\"state\":\"started\",\"pid\":%s,\"statusPath\":\"%s\"}\\n' \"$pid\" \"$status_path\"",
|
|
].join("\n");
|
|
const start = runNodeHostScript(spec, startScript, 55);
|
|
if (!isCommandSuccess(start)) return start;
|
|
const deadline = Date.now() + Math.max(1, timeoutSeconds) * 1000;
|
|
let lastStatus: CommandResult | null = null;
|
|
let payload: Record<string, unknown> = {};
|
|
while (Date.now() <= deadline) {
|
|
runCommand(["sleep", "2"], repoRoot, { timeoutMs: 3_000 });
|
|
const status = runNodeHostScript(spec, `cat ${shellQuote(statusPath)} 2>/dev/null || printf '{"state":"missing","ok":false,"exitCode":127,"stdout":"","stderr":"missing async status"}\\n'`, 55);
|
|
lastStatus = status;
|
|
payload = parseJsonObject(status.stdout.trim());
|
|
if (payload.state === "finished" || payload.ok === true || payload.ok === false) {
|
|
return commandResultFromAsync(spec, payload, statusPath, false);
|
|
}
|
|
}
|
|
const pending = runNodeHostScript(spec, [
|
|
"set +e",
|
|
`status_path=${shellQuote(statusPath)}`,
|
|
`stdout_path=${shellQuote(stdoutPath)}`,
|
|
`stderr_path=${shellQuote(stderrPath)}`,
|
|
`timeout_seconds=${shellQuote(String(timeoutSeconds))}`,
|
|
"python3 - \"$status_path\" \"$stdout_path\" \"$stderr_path\" \"$timeout_seconds\" <<'PY'",
|
|
"import datetime, json, pathlib, subprocess, sys",
|
|
"status_path, stdout_path, stderr_path, timeout_seconds = sys.argv[1:5]",
|
|
"try:",
|
|
" payload = json.load(open(status_path))",
|
|
"except Exception:",
|
|
" payload = {'state': 'missing', 'ok': False, 'exitCode': 127}",
|
|
"def tail(path):",
|
|
" try:",
|
|
" return pathlib.Path(path).read_text(errors='replace')[-12000:]",
|
|
" except FileNotFoundError:",
|
|
" return ''",
|
|
"pid = payload.get('pid')",
|
|
"pid_alive = None",
|
|
"if isinstance(pid, int) and pid > 0:",
|
|
" pid_alive = subprocess.run(['kill', '-0', str(pid)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL).returncode == 0",
|
|
"payload['stdout'] = tail(stdout_path)",
|
|
"stderr = tail(stderr_path)",
|
|
"pending_note = f\"remote async command still pending after {timeout_seconds}s; statusPath={status_path}; pidAlive={pid_alive}\"",
|
|
"payload['stderr'] = '\\n'.join([item for item in [pending_note, stderr] if item])",
|
|
"payload['timedOutAt'] = datetime.datetime.utcnow().isoformat() + 'Z'",
|
|
"payload['pidAlive'] = pid_alive",
|
|
"json.dump(payload, sys.stdout, indent=2)",
|
|
"PY",
|
|
].join("\n"), 55);
|
|
payload = parseJsonObject(pending.stdout.trim());
|
|
return {
|
|
command: [transPath(), spec.nodeRoute, "sh", "--", "<remote async script>"],
|
|
cwd: repoRoot,
|
|
exitCode: 124,
|
|
stdout: typeof payload.stdout === "string" ? payload.stdout : pending.stdout,
|
|
stderr: [
|
|
`remote async command pending after ${timeoutSeconds}s; statusPath=${statusPath}`,
|
|
typeof payload.stderr === "string" ? payload.stderr : "",
|
|
lastStatus?.stderr.trim() ?? "",
|
|
pending.stderr.trim(),
|
|
].filter(Boolean).join("\n"),
|
|
signal: null,
|
|
timedOut: true,
|
|
};
|
|
}
|