diff --git a/.runtime/portal/RUNBOOK.txt b/.runtime/portal/RUNBOOK.txt index fbf3e02..228c714 100644 --- a/.runtime/portal/RUNBOOK.txt +++ b/.runtime/portal/RUNBOOK.txt @@ -38,6 +38,8 @@ Streaming runtime operations: node scripts/runtime-worker-drain.mjs status node scripts/runtime-worker-drain.mjs reconcile node scripts/runtime-worker-drain.mjs reconcile --apply + node scripts/runtime-worker-metrics.mjs status + node scripts/runtime-worker-metrics.mjs sample node scripts/runtime-worker-drain.mjs drain goosed-3 node scripts/runtime-worker-drain.mjs undrain goosed-3 node scripts/check-tool-runtime.mjs diff --git a/.runtime/portal/scripts/runtime-worker-drain.mjs b/.runtime/portal/scripts/runtime-worker-drain.mjs index 1317723..6bfa659 100755 --- a/.runtime/portal/scripts/runtime-worker-drain.mjs +++ b/.runtime/portal/scripts/runtime-worker-drain.mjs @@ -96,7 +96,11 @@ for (const id of [...new Set(workers)].sort()) { activeStreams > 0 && lastStreamStartedAt && staleAgeMs > staleMs && - (!lastStreamEndedAt || lastStreamEndedAt < lastStreamStartedAt), + ( + !lastStreamEndedAt || + lastStreamEndedAt >= lastStreamStartedAt || + now - lastStreamEndedAt > staleMs + ), ); if (action === 'reconcile' && stale && apply) { await client diff --git a/.runtime/portal/scripts/runtime-worker-metrics.mjs b/.runtime/portal/scripts/runtime-worker-metrics.mjs new file mode 100644 index 0000000..c2cf080 --- /dev/null +++ b/.runtime/portal/scripts/runtime-worker-metrics.mjs @@ -0,0 +1,171 @@ +#!/usr/bin/env node +import fs from 'node:fs'; +import path from 'node:path'; +import { execFileSync } from 'node:child_process'; +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 sh(command) { + return execFileSync('/bin/zsh', ['-lc', command], { encoding: 'utf8' }).trim(); +} + +function parsePercent(value) { + const n = Number(String(value ?? '').replace('%', '').trim()); + return Number.isFinite(n) ? n : 0; +} + +function parseBytes(value) { + const match = String(value ?? '').trim().match(/^([\d.]+)\s*([kmgt]?i?b)?$/i); + if (!match) return 0; + const n = Number(match[1]); + if (!Number.isFinite(n)) return 0; + const unit = (match[2] || 'b').toLowerCase(); + const scale = { + b: 1, + kb: 1e3, + mb: 1e6, + gb: 1e9, + tb: 1e12, + kib: 1024, + mib: 1024 ** 2, + gib: 1024 ** 3, + tib: 1024 ** 4, + }[unit] ?? 1; + return Math.round(n * scale); +} + +function parseMemUsage(value) { + const [used, limit] = String(value ?? '').split('/').map((part) => part.trim()); + return { + usedBytes: parseBytes(used), + limitBytes: parseBytes(limit), + }; +} + +function targetWorkerIds() { + 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, + container: `goosed-prod-${index + 1}`, + })); +} + +function readDockerStats(container) { + try { + const line = sh(`docker stats --no-stream --format '{{json .}}' ${JSON.stringify(container)} 2>/dev/null`); + return line ? JSON.parse(line) : null; + } catch { + return null; + } +} + +function readDockerInspect(container) { + try { + const line = sh(`docker inspect --format '{{json .State}}' ${JSON.stringify(container)} 2>/dev/null`); + return line ? JSON.parse(line) : null; + } catch { + return null; + } +} + +function readContainerFdCount(container) { + try { + return Number(sh(`docker exec ${JSON.stringify(container)} sh -lc 'ls /proc/1/fd 2>/dev/null | wc -l'`)) || 0; + } catch { + return 0; + } +} + +function workerKey(namespace, id, field) { + return [namespace, 'worker', id, field].join(':'); +} + +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] || 'sample'; +const dryRun = process.argv.includes('--dry-run') || action === 'status'; +const fdWarn = Number(process.env.GOOSED_FD_WARN ?? 180); +const now = Date.now(); + +if (!['sample', 'status'].includes(action)) { + console.error('Usage: node scripts/runtime-worker-metrics.mjs [--dry-run]'); + process.exit(2); +} + +const workers = []; +for (const worker of targetWorkerIds()) { + const stats = readDockerStats(worker.container); + const state = readDockerInspect(worker.container); + const fdCount = readContainerFdCount(worker.container); + const mem = parseMemUsage(stats?.MemUsage); + const cpuLoad = parsePercent(stats?.CPUPerc); + const memoryPressure = parsePercent(stats?.MemPerc); + const fdPressure = fdWarn > 0 ? Number((fdCount / fdWarn).toFixed(4)) : 0; + workers.push({ + ...worker, + ok: Boolean(stats && state?.Running), + health: state?.Health?.Status ?? state?.Status ?? null, + hostPid: Number(state?.Pid ?? 0) || null, + cpuLoad, + memoryPressure, + fdPressure, + fdCount, + pids: Number(stats?.PIDs ?? 0), + memUsedBytes: mem.usedBytes, + memLimitBytes: mem.limitBytes, + sampledAt: now, + }); +} + +if (!dryRun) { + const client = createClient({ url: redisUrl }); + client.on('error', (err) => { + console.error(`Redis error: ${err instanceof Error ? err.message : err}`); + }); + await client.connect(); + for (const worker of workers) { + await client + .multi() + .set(workerKey(namespace, worker.id, 'cpu_load'), String(worker.cpuLoad)) + .set(workerKey(namespace, worker.id, 'memory_pressure'), String(worker.memoryPressure)) + .set(workerKey(namespace, worker.id, 'fd_pressure'), String(worker.fdPressure)) + .set(workerKey(namespace, worker.id, 'fd_count'), String(worker.fdCount)) + .set(workerKey(namespace, worker.id, 'container_pids'), String(worker.pids)) + .set(workerKey(namespace, worker.id, 'container_name'), worker.container) + .set(workerKey(namespace, worker.id, 'container_health'), worker.health ?? '') + .set(workerKey(namespace, worker.id, 'container_host_pid'), String(worker.hostPid ?? '')) + .set(workerKey(namespace, worker.id, 'metrics_sampled_at'), String(worker.sampledAt)) + .set(workerKey(namespace, worker.id, 'heartbeat'), String(now), { EX: 90 }) + .exec(); + } + await client.quit(); +} + +console.log(JSON.stringify({ + ok: workers.every((worker) => worker.ok), + action, + dryRun, + namespace, + redisWrites: !dryRun, + workers, +}, null, 2)); + +process.exit(workers.every((worker) => worker.ok) ? 0 : 1); diff --git a/.runtime/portal/server.mjs b/.runtime/portal/server.mjs index 703bf64..e0f0ed4 100644 --- a/.runtime/portal/server.mjs +++ b/.runtime/portal/server.mjs @@ -8710,12 +8710,14 @@ function createRuntimeRouter({ key("worker", workerId, "ewma_first_token_ms"), key("worker", workerId, "error_rate"), key("worker", workerId, "memory_pressure"), + key("worker", workerId, "cpu_load"), + key("worker", workerId, "fd_pressure"), key("worker", workerId, "drain") ]); - if (/^(1|true|yes)$/i.test(String(values[5] ?? ""))) return Number.POSITIVE_INFINITY; - return readNumber(values[0]) * 3 + readNumber(values[1]) + readNumber(values[2]) * 0.01 + readNumber(values[3]) * 5 + readNumber(values[4]) * 2; + if (/^(1|true|yes)$/i.test(String(values[7] ?? ""))) return Number.POSITIVE_INFINITY; + return readNumber(values[0]) * 3 + readNumber(values[1]) + readNumber(values[2]) * 0.01 + readNumber(values[3]) * 5 + readNumber(values[4]) * 2 + readNumber(values[5]) * 2 + readNumber(values[6]) * 2; }; - const workerScoreFromValues = (values = []) => readNumber(values[0]) * 3 + readNumber(values[1]) + readNumber(values[2]) * 0.01 + readNumber(values[3]) * 5 + readNumber(values[4]) * 2; + const workerScoreFromValues = (values = []) => readNumber(values[0]) * 3 + readNumber(values[1]) + readNumber(values[2]) * 0.01 + readNumber(values[3]) * 5 + readNumber(values[4]) * 2 + readNumber(values[12]) * 2 + readNumber(values[13]) * 2; return { async pickTarget(fallbackPick, orderedTargets = targets) { const client = await getClient(); @@ -8783,7 +8785,13 @@ function createRuntimeRouter({ key("worker", workerId, "stream_abort_count"), key("worker", workerId, "stream_error_count"), key("worker", workerId, "last_stream_started_at"), - key("worker", workerId, "last_stream_ended_at") + key("worker", workerId, "last_stream_ended_at"), + key("worker", workerId, "cpu_load"), + key("worker", workerId, "fd_pressure"), + key("worker", workerId, "fd_count"), + key("worker", workerId, "container_pids"), + key("worker", workerId, "container_health"), + key("worker", workerId, "metrics_sampled_at") ]).catch(() => []) : []; workers.push({ id: workerId, @@ -8800,6 +8808,12 @@ function createRuntimeRouter({ streamErrorCount: readNumber(values?.[9]), lastStreamStartedAt: values?.[10] ? Number(values[10]) : null, lastStreamEndedAt: values?.[11] ? Number(values[11]) : null, + cpuLoad: readNumber(values?.[12]), + fdPressure: readNumber(values?.[13]), + fdCount: readNumber(values?.[14]), + containerPids: readNumber(values?.[15]), + containerHealth: values?.[16] ?? null, + metricsSampledAt: values?.[17] ? Number(values[17]) : null, score: workerScoreFromValues(values) }); } diff --git a/scripts/build-portal-runtime.mjs b/scripts/build-portal-runtime.mjs index 0633174..b2850c5 100755 --- a/scripts/build-portal-runtime.mjs +++ b/scripts/build-portal-runtime.mjs @@ -275,6 +275,10 @@ async function writeMetadata() { path.join(root, 'scripts', 'runtime-worker-drain.mjs'), path.join(runtimeRoot, 'scripts', 'runtime-worker-drain.mjs'), ); + await fs.copyFile( + path.join(root, 'scripts', 'runtime-worker-metrics.mjs'), + path.join(runtimeRoot, 'scripts', 'runtime-worker-metrics.mjs'), + ); await fs.copyFile( path.join(root, 'scripts', 'check-tool-runtime.mjs'), path.join(runtimeRoot, 'scripts', 'check-tool-runtime.mjs'), @@ -326,6 +330,8 @@ async function writeMetadata() { ' node scripts/runtime-worker-drain.mjs status', ' node scripts/runtime-worker-drain.mjs reconcile', ' node scripts/runtime-worker-drain.mjs reconcile --apply', + ' node scripts/runtime-worker-metrics.mjs status', + ' node scripts/runtime-worker-metrics.mjs sample', ' node scripts/runtime-worker-drain.mjs drain goosed-3', ' node scripts/runtime-worker-drain.mjs undrain goosed-3', ' node scripts/check-tool-runtime.mjs', diff --git a/scripts/runtime-worker-drain.mjs b/scripts/runtime-worker-drain.mjs index 1317723..6bfa659 100755 --- a/scripts/runtime-worker-drain.mjs +++ b/scripts/runtime-worker-drain.mjs @@ -96,7 +96,11 @@ for (const id of [...new Set(workers)].sort()) { activeStreams > 0 && lastStreamStartedAt && staleAgeMs > staleMs && - (!lastStreamEndedAt || lastStreamEndedAt < lastStreamStartedAt), + ( + !lastStreamEndedAt || + lastStreamEndedAt >= lastStreamStartedAt || + now - lastStreamEndedAt > staleMs + ), ); if (action === 'reconcile' && stale && apply) { await client diff --git a/scripts/runtime-worker-metrics.mjs b/scripts/runtime-worker-metrics.mjs new file mode 100644 index 0000000..c2cf080 --- /dev/null +++ b/scripts/runtime-worker-metrics.mjs @@ -0,0 +1,171 @@ +#!/usr/bin/env node +import fs from 'node:fs'; +import path from 'node:path'; +import { execFileSync } from 'node:child_process'; +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 sh(command) { + return execFileSync('/bin/zsh', ['-lc', command], { encoding: 'utf8' }).trim(); +} + +function parsePercent(value) { + const n = Number(String(value ?? '').replace('%', '').trim()); + return Number.isFinite(n) ? n : 0; +} + +function parseBytes(value) { + const match = String(value ?? '').trim().match(/^([\d.]+)\s*([kmgt]?i?b)?$/i); + if (!match) return 0; + const n = Number(match[1]); + if (!Number.isFinite(n)) return 0; + const unit = (match[2] || 'b').toLowerCase(); + const scale = { + b: 1, + kb: 1e3, + mb: 1e6, + gb: 1e9, + tb: 1e12, + kib: 1024, + mib: 1024 ** 2, + gib: 1024 ** 3, + tib: 1024 ** 4, + }[unit] ?? 1; + return Math.round(n * scale); +} + +function parseMemUsage(value) { + const [used, limit] = String(value ?? '').split('/').map((part) => part.trim()); + return { + usedBytes: parseBytes(used), + limitBytes: parseBytes(limit), + }; +} + +function targetWorkerIds() { + 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, + container: `goosed-prod-${index + 1}`, + })); +} + +function readDockerStats(container) { + try { + const line = sh(`docker stats --no-stream --format '{{json .}}' ${JSON.stringify(container)} 2>/dev/null`); + return line ? JSON.parse(line) : null; + } catch { + return null; + } +} + +function readDockerInspect(container) { + try { + const line = sh(`docker inspect --format '{{json .State}}' ${JSON.stringify(container)} 2>/dev/null`); + return line ? JSON.parse(line) : null; + } catch { + return null; + } +} + +function readContainerFdCount(container) { + try { + return Number(sh(`docker exec ${JSON.stringify(container)} sh -lc 'ls /proc/1/fd 2>/dev/null | wc -l'`)) || 0; + } catch { + return 0; + } +} + +function workerKey(namespace, id, field) { + return [namespace, 'worker', id, field].join(':'); +} + +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] || 'sample'; +const dryRun = process.argv.includes('--dry-run') || action === 'status'; +const fdWarn = Number(process.env.GOOSED_FD_WARN ?? 180); +const now = Date.now(); + +if (!['sample', 'status'].includes(action)) { + console.error('Usage: node scripts/runtime-worker-metrics.mjs [--dry-run]'); + process.exit(2); +} + +const workers = []; +for (const worker of targetWorkerIds()) { + const stats = readDockerStats(worker.container); + const state = readDockerInspect(worker.container); + const fdCount = readContainerFdCount(worker.container); + const mem = parseMemUsage(stats?.MemUsage); + const cpuLoad = parsePercent(stats?.CPUPerc); + const memoryPressure = parsePercent(stats?.MemPerc); + const fdPressure = fdWarn > 0 ? Number((fdCount / fdWarn).toFixed(4)) : 0; + workers.push({ + ...worker, + ok: Boolean(stats && state?.Running), + health: state?.Health?.Status ?? state?.Status ?? null, + hostPid: Number(state?.Pid ?? 0) || null, + cpuLoad, + memoryPressure, + fdPressure, + fdCount, + pids: Number(stats?.PIDs ?? 0), + memUsedBytes: mem.usedBytes, + memLimitBytes: mem.limitBytes, + sampledAt: now, + }); +} + +if (!dryRun) { + const client = createClient({ url: redisUrl }); + client.on('error', (err) => { + console.error(`Redis error: ${err instanceof Error ? err.message : err}`); + }); + await client.connect(); + for (const worker of workers) { + await client + .multi() + .set(workerKey(namespace, worker.id, 'cpu_load'), String(worker.cpuLoad)) + .set(workerKey(namespace, worker.id, 'memory_pressure'), String(worker.memoryPressure)) + .set(workerKey(namespace, worker.id, 'fd_pressure'), String(worker.fdPressure)) + .set(workerKey(namespace, worker.id, 'fd_count'), String(worker.fdCount)) + .set(workerKey(namespace, worker.id, 'container_pids'), String(worker.pids)) + .set(workerKey(namespace, worker.id, 'container_name'), worker.container) + .set(workerKey(namespace, worker.id, 'container_health'), worker.health ?? '') + .set(workerKey(namespace, worker.id, 'container_host_pid'), String(worker.hostPid ?? '')) + .set(workerKey(namespace, worker.id, 'metrics_sampled_at'), String(worker.sampledAt)) + .set(workerKey(namespace, worker.id, 'heartbeat'), String(now), { EX: 90 }) + .exec(); + } + await client.quit(); +} + +console.log(JSON.stringify({ + ok: workers.every((worker) => worker.ok), + action, + dryRun, + namespace, + redisWrites: !dryRun, + workers, +}, null, 2)); + +process.exit(workers.every((worker) => worker.ok) ? 0 : 1); diff --git a/tkmind-proxy.mjs b/tkmind-proxy.mjs index 25a0b15..f063abd 100644 --- a/tkmind-proxy.mjs +++ b/tkmind-proxy.mjs @@ -132,15 +132,19 @@ function createRuntimeRouter({ key('worker', workerId, 'ewma_first_token_ms'), key('worker', workerId, 'error_rate'), key('worker', workerId, 'memory_pressure'), + key('worker', workerId, 'cpu_load'), + key('worker', workerId, 'fd_pressure'), key('worker', workerId, 'drain'), ]); - if (/^(1|true|yes)$/i.test(String(values[5] ?? ''))) return Number.POSITIVE_INFINITY; + if (/^(1|true|yes)$/i.test(String(values[7] ?? ''))) return Number.POSITIVE_INFINITY; return ( readNumber(values[0]) * 3 + readNumber(values[1]) + readNumber(values[2]) * 0.01 + readNumber(values[3]) * 5 + - readNumber(values[4]) * 2 + readNumber(values[4]) * 2 + + readNumber(values[5]) * 2 + + readNumber(values[6]) * 2 ); }; const workerScoreFromValues = (values = []) => ( @@ -148,7 +152,9 @@ function createRuntimeRouter({ readNumber(values[1]) + readNumber(values[2]) * 0.01 + readNumber(values[3]) * 5 + - readNumber(values[4]) * 2 + readNumber(values[4]) * 2 + + readNumber(values[12]) * 2 + + readNumber(values[13]) * 2 ); return { async pickTarget(fallbackPick, orderedTargets = targets) { @@ -233,6 +239,12 @@ function createRuntimeRouter({ key('worker', workerId, 'stream_error_count'), key('worker', workerId, 'last_stream_started_at'), key('worker', workerId, 'last_stream_ended_at'), + key('worker', workerId, 'cpu_load'), + key('worker', workerId, 'fd_pressure'), + key('worker', workerId, 'fd_count'), + key('worker', workerId, 'container_pids'), + key('worker', workerId, 'container_health'), + key('worker', workerId, 'metrics_sampled_at'), ]) .catch(() => []) : []; @@ -251,6 +263,12 @@ function createRuntimeRouter({ streamErrorCount: readNumber(values?.[9]), lastStreamStartedAt: values?.[10] ? Number(values[10]) : null, lastStreamEndedAt: values?.[11] ? Number(values[11]) : null, + cpuLoad: readNumber(values?.[12]), + fdPressure: readNumber(values?.[13]), + fdCount: readNumber(values?.[14]), + containerPids: readNumber(values?.[15]), + containerHealth: values?.[16] ?? null, + metricsSampledAt: values?.[17] ? Number(values[17]) : null, score: workerScoreFromValues(values), }); }