From e24e6ca4a72b4cc16a2074d205727f6d131cff0d Mon Sep 17 00:00:00 2001 From: John Date: Thu, 2 Jul 2026 09:03:30 +0800 Subject: [PATCH] feat: add runtime worker heartbeat sidecar --- scripts/build-portal-runtime.mjs | 13 ++ scripts/install-runtime-heartbeat-agent.sh | 76 ++++++++++ scripts/runtime-slo-report.mjs | 16 +- scripts/runtime-worker-heartbeat.mjs | 164 +++++++++++++++++++++ tkmind-proxy.mjs | 10 ++ 5 files changed, 277 insertions(+), 2 deletions(-) create mode 100644 scripts/install-runtime-heartbeat-agent.sh create mode 100644 scripts/runtime-worker-heartbeat.mjs diff --git a/scripts/build-portal-runtime.mjs b/scripts/build-portal-runtime.mjs index 4eba621..38da84b 100755 --- a/scripts/build-portal-runtime.mjs +++ b/scripts/build-portal-runtime.mjs @@ -299,6 +299,10 @@ async function writeMetadata() { path.join(root, 'scripts', 'runtime-worker-metrics.mjs'), path.join(runtimeRoot, 'scripts', 'runtime-worker-metrics.mjs'), ); + await fs.copyFile( + path.join(root, 'scripts', 'runtime-worker-heartbeat.mjs'), + path.join(runtimeRoot, 'scripts', 'runtime-worker-heartbeat.mjs'), + ); await fs.copyFile( path.join(root, 'scripts', 'runtime-slo-report.mjs'), path.join(runtimeRoot, 'scripts', 'runtime-slo-report.mjs'), @@ -307,6 +311,10 @@ async function writeMetadata() { path.join(root, 'scripts', 'install-runtime-metrics-agent.sh'), path.join(runtimeRoot, 'scripts', 'install-runtime-metrics-agent.sh'), ); + await fs.copyFile( + path.join(root, 'scripts', 'install-runtime-heartbeat-agent.sh'), + path.join(runtimeRoot, 'scripts', 'install-runtime-heartbeat-agent.sh'), + ); await fs.copyFile( path.join(root, 'scripts', 'install-runtime-slo-report-agent.sh'), path.join(runtimeRoot, 'scripts', 'install-runtime-slo-report-agent.sh'), @@ -365,6 +373,9 @@ async function writeMetadata() { ' node scripts/runtime-worker-metrics.mjs status', ' node scripts/runtime-worker-metrics.mjs sample', ' bash scripts/install-runtime-metrics-agent.sh', + ' node scripts/runtime-worker-heartbeat.mjs once', + ' node scripts/runtime-worker-heartbeat.mjs serve', + ' bash scripts/install-runtime-heartbeat-agent.sh', ' node scripts/runtime-slo-report.mjs', ' node scripts/runtime-slo-report.mjs --write-report', ' node scripts/runtime-slo-report.mjs --write-report --prune --retention-days 30', @@ -397,9 +408,11 @@ async function main() { await fs.chmod(path.join(runtimeRoot, 'scripts', 'check-stream-runtime.mjs'), 0o755); await fs.chmod(path.join(runtimeRoot, 'scripts', 'runtime-worker-drain.mjs'), 0o755); await fs.chmod(path.join(runtimeRoot, 'scripts', 'runtime-worker-metrics.mjs'), 0o755); + await fs.chmod(path.join(runtimeRoot, 'scripts', 'runtime-worker-heartbeat.mjs'), 0o755); await fs.chmod(path.join(runtimeRoot, 'scripts', 'runtime-slo-report.mjs'), 0o755); await fs.chmod(path.join(runtimeRoot, 'scripts', 'agent-run-worker.mjs'), 0o755); await fs.chmod(path.join(runtimeRoot, 'scripts', 'install-runtime-metrics-agent.sh'), 0o755); + await fs.chmod(path.join(runtimeRoot, 'scripts', 'install-runtime-heartbeat-agent.sh'), 0o755); await fs.chmod(path.join(runtimeRoot, 'scripts', 'install-runtime-slo-report-agent.sh'), 0o755); await fs.chmod(path.join(runtimeRoot, 'scripts', 'check-tool-runtime.mjs'), 0o755); await fs.chmod(path.join(runtimeRoot, 'scripts', 'memind-portal-tunnel.sh'), 0o755); diff --git a/scripts/install-runtime-heartbeat-agent.sh b/scripts/install-runtime-heartbeat-agent.sh new file mode 100644 index 0000000..178d36e --- /dev/null +++ b/scripts/install-runtime-heartbeat-agent.sh @@ -0,0 +1,76 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT="$(cd "$(dirname "$0")/.." && pwd)" +NODE_BIN="${NODE_BIN:-/opt/homebrew/opt/node@24/bin/node}" +SCRIPT="${MEMIND_RUNTIME_HEARTBEAT_SCRIPT:-$ROOT/scripts/runtime-worker-heartbeat.mjs}" +LABEL="${MEMIND_RUNTIME_HEARTBEAT_LABEL:-cn.tkmind.memind-runtime-heartbeat}" +PLIST="$HOME/Library/LaunchAgents/${LABEL}.plist" +LOG="$HOME/Library/Logs/memind-runtime-heartbeat.log" +GUI="gui/$(id -u)" +INTERVAL_MS="${MEMIND_RUNTIME_HEARTBEAT_INTERVAL_MS:-15000}" +TIMEOUT_MS="${MEMIND_RUNTIME_HEARTBEAT_TIMEOUT_MS:-5000}" +TTL_MS="${MEMIND_RUNTIME_HEARTBEAT_TTL_MS:-45000}" + +mkdir -p "$HOME/Library/LaunchAgents" "$HOME/Library/Logs" + +if [[ ! -x "$NODE_BIN" ]]; then + NODE_BIN="$(command -v node)" +fi +if [[ ! -f "$SCRIPT" ]]; then + echo "runtime heartbeat script not found: $SCRIPT" >&2 + exit 1 +fi + +cat > "$PLIST" < + + + + Label + $LABEL + ProgramArguments + + $NODE_BIN + $SCRIPT + serve + + WorkingDirectory + $ROOT + RunAtLoad + + KeepAlive + + ThrottleInterval + 10 + StandardOutPath + $LOG + StandardErrorPath + $LOG + EnvironmentVariables + + PATH + /opt/homebrew/bin:/opt/homebrew/opt/node@24/bin:/usr/local/bin:/usr/bin:/bin + MEMIND_RUNTIME_HEARTBEAT_INTERVAL_MS + $INTERVAL_MS + MEMIND_RUNTIME_HEARTBEAT_TIMEOUT_MS + $TIMEOUT_MS + MEMIND_RUNTIME_HEARTBEAT_TTL_MS + $TTL_MS + + + +EOF + +plutil -lint "$PLIST" +launchctl bootout "$GUI/$LABEL" 2>/dev/null || true +launchctl bootstrap "$GUI" "$PLIST" +launchctl enable "$GUI/$LABEL" +launchctl kickstart -k "$GUI/$LABEL" + +echo "installed $PLIST" +echo "script: $SCRIPT" +echo "interval_ms: $INTERVAL_MS" +echo "timeout_ms: $TIMEOUT_MS" +echo "ttl_ms: $TTL_MS" +echo "log: $LOG" diff --git a/scripts/runtime-slo-report.mjs b/scripts/runtime-slo-report.mjs index da34e2d..6c26218 100755 --- a/scripts/runtime-slo-report.mjs +++ b/scripts/runtime-slo-report.mjs @@ -225,6 +225,7 @@ async function readRedisSummary(namespace, redisUrl) { function summarizeRuntime(runtimeJson) { const workers = runtimeJson?.router?.workers ?? []; const staleMetricMs = 2 * 60 * 1000; + const staleHeartbeatMs = Number(process.env.MEMIND_RUNTIME_HEARTBEAT_STALE_MS || 60 * 1000); const now = Date.now(); return { routerEnabled: Boolean(runtimeJson?.router?.enabled), @@ -249,6 +250,14 @@ function summarizeRuntime(runtimeJson) { fdPressure: worker.fdPressure, fdCount: worker.fdCount, containerHealth: worker.containerHealth, + heartbeat: worker.heartbeat, + heartbeatAgeMs: worker.heartbeat ? now - worker.heartbeat : null, + heartbeatFresh: worker.heartbeat ? now - worker.heartbeat <= staleHeartbeatMs : false, + heartbeatSource: worker.heartbeatSource ?? null, + heartbeatOk: worker.heartbeatOk ?? null, + heartbeatStatusCode: worker.heartbeatStatusCode ?? null, + heartbeatLatencyMs: worker.heartbeatLatencyMs ?? null, + heartbeatError: worker.heartbeatError ?? null, metricsAgeMs: worker.metricsSampledAt ? now - worker.metricsSampledAt : null, metricsFresh: worker.metricsSampledAt ? now - worker.metricsSampledAt <= staleMetricMs : false, score: worker.score, @@ -304,6 +313,8 @@ if (!runtime.ok) failures.push('runtime_status_unavailable'); if (runtimeSummary && !runtimeSummary.routerEnabled) failures.push('router_disabled'); for (const worker of runtimeSummary?.workers ?? []) { if (!worker.healthy) failures.push(`${worker.id}_target_unhealthy`); + if (!worker.heartbeatFresh) failures.push(`${worker.id}_heartbeat_stale`); + if (worker.heartbeatOk === false) failures.push(`${worker.id}_heartbeat_unhealthy`); if (worker.containerHealth && worker.containerHealth !== 'healthy') failures.push(`${worker.id}_container_${worker.containerHealth}`); if (!worker.metricsFresh) failures.push(`${worker.id}_metrics_stale`); if (worker.streamErrorCount > 0) failures.push(`${worker.id}_stream_errors_${worker.streamErrorCount}`); @@ -361,8 +372,8 @@ function markdownReport(payload) { '', '## Workers', '', - '| worker | healthy | active | errors | first-token 5m p50/p95 | first-token 1h p50/p95 | metricsFresh | score |', - '| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: |', + '| worker | healthy | active | errors | first-token 5m p50/p95 | first-token 1h p50/p95 | heartbeat | metricsFresh | score |', + '| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |', ]; for (const worker of workers) { lines.push([ @@ -372,6 +383,7 @@ function markdownReport(payload) { worker.streamErrorCount, `${worker.firstToken5m?.p50Ms ?? 0}/${worker.firstToken5m?.p95Ms ?? 0}`, `${worker.firstToken1h?.p50Ms ?? 0}/${worker.firstToken1h?.p95Ms ?? 0}`, + `${worker.heartbeatSource ?? 'none'}:${worker.heartbeatFresh}`, worker.metricsFresh, worker.score, ].join(' | ').replace(/^/, '| ').replace(/$/, ' |')); diff --git a/scripts/runtime-worker-heartbeat.mjs b/scripts/runtime-worker-heartbeat.mjs new file mode 100644 index 0000000..9de4194 --- /dev/null +++ b/scripts/runtime-worker-heartbeat.mjs @@ -0,0 +1,164 @@ +#!/usr/bin/env node +import fs from 'node:fs'; +import path from 'node:path'; +import { Agent, fetch as undiciFetch } from 'undici'; +import { createClient } from 'redis'; + +function loadEnvFile(filePath) { + if (!fs.existsSync(filePath)) return; + for (const line of fs.readFileSync(filePath, 'utf8').split('\n')) { + const trimmed = line.trim(); + if (!trimmed || trimmed.startsWith('#')) continue; + const idx = trimmed.indexOf('='); + if (idx < 0) continue; + const key = trimmed.slice(0, idx).trim(); + const value = trimmed.slice(idx + 1).trim(); + if (!process.env[key]) process.env[key] = value; + } +} + +function positiveInteger(value, fallback) { + const n = Number(value); + if (!Number.isFinite(n) || n <= 0) return fallback; + return Math.floor(n); +} + +function targetWorkers() { + const targets = (process.env.TKMIND_API_TARGETS || process.env.TKMIND_API_TARGET || '') + .split(',') + .map((value) => value.trim()) + .filter(Boolean); + return targets.map((target, index) => ({ + id: `goosed-${index + 1}`, + target, + })); +} + +function workerKey(namespace, id, field) { + return [namespace, 'worker', id, field].join(':'); +} + +async function probeTarget(target, timeoutMs) { + const startedAt = Date.now(); + const controller = new AbortController(); + const timeout = setTimeout(() => controller.abort(), timeoutMs); + const dispatcher = new Agent({ connect: { rejectUnauthorized: false } }); + try { + const url = new URL('/status', target.endsWith('/') ? target : `${target}/`).toString(); + const res = await undiciFetch(url, { + signal: controller.signal, + dispatcher: url.startsWith('https://127.0.0.1') ? dispatcher : undefined, + headers: process.env.TKMIND_SERVER__SECRET_KEY + ? { 'X-Secret-Key': process.env.TKMIND_SERVER__SECRET_KEY } + : undefined, + }); + const body = (await res.text()).trim().slice(0, 80); + return { + ok: res.ok && body === 'ok', + statusCode: res.status, + body, + latencyMs: Date.now() - startedAt, + error: null, + }; + } catch (err) { + return { + ok: false, + statusCode: 0, + body: '', + latencyMs: Date.now() - startedAt, + error: err instanceof Error ? err.message : String(err), + }; + } finally { + clearTimeout(timeout); + dispatcher.close(); + } +} + +async function writeHeartbeats(client, namespace, workers, options) { + const now = Date.now(); + const results = []; + for (const worker of workers) { + const probe = await probeTarget(worker.target, options.timeoutMs); + const ttlSeconds = Math.max(5, Math.ceil(options.ttlMs / 1000)); + const multi = client + .multi() + .set(workerKey(namespace, worker.id, 'heartbeat'), String(now), { EX: ttlSeconds }) + .set(workerKey(namespace, worker.id, 'heartbeat_at'), String(now), { EX: ttlSeconds }) + .set(workerKey(namespace, worker.id, 'heartbeat_source'), 'sidecar', { EX: ttlSeconds }) + .set(workerKey(namespace, worker.id, 'heartbeat_target'), worker.target, { EX: ttlSeconds }) + .set(workerKey(namespace, worker.id, 'heartbeat_ok'), probe.ok ? '1' : '0', { EX: ttlSeconds }) + .set(workerKey(namespace, worker.id, 'heartbeat_status_code'), String(probe.statusCode), { EX: ttlSeconds }) + .set(workerKey(namespace, worker.id, 'heartbeat_latency_ms'), String(probe.latencyMs), { EX: ttlSeconds }); + if (probe.error) { + multi.set(workerKey(namespace, worker.id, 'heartbeat_error'), probe.error, { EX: ttlSeconds }); + } else { + multi.del(workerKey(namespace, worker.id, 'heartbeat_error')); + } + await multi.exec(); + results.push({ ...worker, ...probe, heartbeatAt: now }); + } + return results; +} + +loadEnvFile(process.env.MEMIND_ENV_FILE || path.join(process.cwd(), '.env')); + +const redisUrl = process.env.MEMIND_RUNTIME_REDIS_URL || 'redis://127.0.0.1:6379/0'; +const namespace = process.env.MEMIND_RUNTIME_REDIS_NAMESPACE || 'memind:runtime'; +const action = process.argv[2] || 'once'; +const intervalMs = positiveInteger(process.env.MEMIND_RUNTIME_HEARTBEAT_INTERVAL_MS, 15_000); +const timeoutMs = positiveInteger(process.env.MEMIND_RUNTIME_HEARTBEAT_TIMEOUT_MS, 5_000); +const ttlMs = positiveInteger(process.env.MEMIND_RUNTIME_HEARTBEAT_TTL_MS, Math.max(45_000, intervalMs * 3)); + +if (!['once', 'serve'].includes(action)) { + console.error('Usage: node scripts/runtime-worker-heartbeat.mjs '); + process.exit(2); +} + +const workers = targetWorkers(); +if (workers.length === 0) { + console.error('No TKMIND_API_TARGETS configured'); + process.exit(2); +} + +const client = createClient({ url: redisUrl }); +client.on('error', (err) => { + console.error(`Redis error: ${err instanceof Error ? err.message : err}`); +}); +await client.connect(); + +let stopping = false; +const stop = async () => { + stopping = true; + await client.quit().catch(() => {}); +}; +process.on('SIGTERM', () => void stop().finally(() => process.exit(0))); +process.on('SIGINT', () => void stop().finally(() => process.exit(0))); + +async function tick() { + const result = await writeHeartbeats(client, namespace, workers, { timeoutMs, ttlMs }); + const payload = { + ok: result.every((worker) => worker.ok), + action, + namespace, + intervalMs: action === 'serve' ? intervalMs : undefined, + timeoutMs, + ttlMs, + workers: result, + }; + console.log(JSON.stringify(payload)); + return payload; +} + +if (action === 'once') { + const payload = await tick(); + await client.quit(); + process.exit(payload.ok ? 0 : 1); +} + +await tick(); +while (!stopping) { + await new Promise((resolve) => setTimeout(resolve, intervalMs)); + if (!stopping) await tick().catch((err) => { + console.error(err instanceof Error ? err.message : String(err)); + }); +} diff --git a/tkmind-proxy.mjs b/tkmind-proxy.mjs index 2cabf0a..67550b9 100644 --- a/tkmind-proxy.mjs +++ b/tkmind-proxy.mjs @@ -312,6 +312,11 @@ function createRuntimeRouter({ key('worker', workerId, 'last_first_token_ms'), key('worker', workerId, 'last_first_token_at'), key('worker', workerId, 'first_token_count'), + key('worker', workerId, 'heartbeat_source'), + key('worker', workerId, 'heartbeat_ok'), + key('worker', workerId, 'heartbeat_status_code'), + key('worker', workerId, 'heartbeat_latency_ms'), + key('worker', workerId, 'heartbeat_error'), ]) .catch(() => []) : []; @@ -341,6 +346,11 @@ function createRuntimeRouter({ lastFirstTokenMs: readNumber(values?.[18]), lastFirstTokenAt: values?.[19] ? Number(values[19]) : null, firstTokenCount: readNumber(values?.[20]), + heartbeatSource: values?.[21] ?? null, + heartbeatOk: values?.[22] == null ? null : /^(1|true|yes)$/i.test(String(values[22])), + heartbeatStatusCode: readNumber(values?.[23]), + heartbeatLatencyMs: readNumber(values?.[24]), + heartbeatError: values?.[25] || null, firstToken5m, firstToken1h, score: workerScoreFromValues(values),