From 12f74eb44e106c4e93f33b56d440ee1ee368daf8 Mon Sep 17 00:00:00 2001 From: John Date: Thu, 2 Jul 2026 08:06:58 +0800 Subject: [PATCH] feat: add first token latency window stats --- .runtime/portal/scripts/agent-run-worker.mjs | 42 ++++++++++++++- .../portal/scripts/runtime-slo-report.mjs | 2 + .runtime/portal/server.mjs | 42 ++++++++++++++- scripts/runtime-slo-report.mjs | 2 + tkmind-proxy.mjs | 51 ++++++++++++++++++- 5 files changed, 136 insertions(+), 3 deletions(-) diff --git a/.runtime/portal/scripts/agent-run-worker.mjs b/.runtime/portal/scripts/agent-run-worker.mjs index 0ca5127..20fd3ad 100755 --- a/.runtime/portal/scripts/agent-run-worker.mjs +++ b/.runtime/portal/scripts/agent-run-worker.mjs @@ -2275,6 +2275,32 @@ function createRuntimeRouter({ const n = Number(value ?? 0); return Number.isFinite(n) ? n : 0; }; + const parseFirstTokenSample = (value) => { + const parts = String(value ?? "").split(":"); + return readNumber(parts[1]); + }; + const percentile = (values, pct) => { + const sorted = values.map((value) => readNumber(value)).filter((value) => value > 0).sort((a, b) => a - b); + if (sorted.length === 0) return 0; + const idx = Math.min(sorted.length - 1, Math.max(0, Math.ceil(pct / 100 * sorted.length) - 1)); + return sorted[idx]; + }; + const firstTokenWindowStats = async (client, workerId, windowMs) => { + if (!client) return { count: 0, p50Ms: 0, p95Ms: 0 }; + const now = Date.now(); + const samples = await client.sendCommand([ + "ZRANGEBYSCORE", + key("worker", workerId, "first_token_samples"), + String(now - windowMs), + String(now) + ]).catch(() => []); + const values = samples.map(parseFirstTokenSample).filter((value) => value > 0); + return { + count: values.length, + p50Ms: percentile(values, 50), + p95Ms: percentile(values, 95) + }; + }; const scoreWorker = async (client, target) => { const workerId = workerIdForTarget(target); const values = await client.mGet([ @@ -2347,10 +2373,20 @@ function createRuntimeRouter({ const workerId = workerIdForTarget(target); const sample = Math.max(0, Math.round(Number(latencyMs) || 0)); const ewmaKey = key("worker", workerId, "ewma_first_token_ms"); + const samplesKey = key("worker", workerId, "first_token_samples"); const prev = readNumber(await client.get(ewmaKey).catch(() => null)); const next = prev > 0 ? Math.round(prev * 0.8 + sample * 0.2) : sample; - const now = String(Date.now()); + const nowMs2 = Date.now(); + const now = String(nowMs2); await client.multi().set(ewmaKey, String(next)).set(key("worker", workerId, "last_first_token_ms"), String(sample)).set(key("worker", workerId, "last_first_token_at"), now).incr(key("worker", workerId, "first_token_count")).set(key("worker", workerId, "heartbeat"), now, { EX: 30 }).exec().catch(() => null); + await client.sendCommand([ + "ZADD", + samplesKey, + String(nowMs2), + `${now}:${sample}:${Math.random().toString(36).slice(2)}` + ]).catch(() => null); + await client.sendCommand(["ZREMRANGEBYSCORE", samplesKey, "0", String(nowMs2 - 60 * 60 * 1e3)]).catch(() => null); + await client.sendCommand(["EXPIRE", samplesKey, String(2 * 60 * 60)]).catch(() => null); }, async getStatus() { const client = await getClient(); @@ -2380,6 +2416,8 @@ function createRuntimeRouter({ key("worker", workerId, "last_first_token_at"), key("worker", workerId, "first_token_count") ]).catch(() => []) : []; + const firstToken5m = await firstTokenWindowStats(client, workerId, 5 * 60 * 1e3); + const firstToken1h = await firstTokenWindowStats(client, workerId, 60 * 60 * 1e3); workers.push({ id: workerId, target, @@ -2404,6 +2442,8 @@ function createRuntimeRouter({ lastFirstTokenMs: readNumber(values?.[18]), lastFirstTokenAt: values?.[19] ? Number(values[19]) : null, firstTokenCount: readNumber(values?.[20]), + firstToken5m, + firstToken1h, score: workerScoreFromValues(values) }); } diff --git a/.runtime/portal/scripts/runtime-slo-report.mjs b/.runtime/portal/scripts/runtime-slo-report.mjs index dd52c08..88ec96f 100755 --- a/.runtime/portal/scripts/runtime-slo-report.mjs +++ b/.runtime/portal/scripts/runtime-slo-report.mjs @@ -160,6 +160,8 @@ function summarizeRuntime(runtimeJson) { lastFirstTokenMs: worker.lastFirstTokenMs, lastFirstTokenAt: worker.lastFirstTokenAt, firstTokenCount: worker.firstTokenCount, + firstToken5m: worker.firstToken5m ?? null, + firstToken1h: worker.firstToken1h ?? null, cpuLoad: worker.cpuLoad, memoryPressure: worker.memoryPressure, fdPressure: worker.fdPressure, diff --git a/.runtime/portal/server.mjs b/.runtime/portal/server.mjs index 0610c15..24c8632 100644 --- a/.runtime/portal/server.mjs +++ b/.runtime/portal/server.mjs @@ -8820,6 +8820,32 @@ function createRuntimeRouter({ const n = Number(value ?? 0); return Number.isFinite(n) ? n : 0; }; + const parseFirstTokenSample = (value) => { + const parts = String(value ?? "").split(":"); + return readNumber(parts[1]); + }; + const percentile = (values, pct) => { + const sorted = values.map((value) => readNumber(value)).filter((value) => value > 0).sort((a, b) => a - b); + if (sorted.length === 0) return 0; + const idx = Math.min(sorted.length - 1, Math.max(0, Math.ceil(pct / 100 * sorted.length) - 1)); + return sorted[idx]; + }; + const firstTokenWindowStats = async (client, workerId, windowMs) => { + if (!client) return { count: 0, p50Ms: 0, p95Ms: 0 }; + const now = Date.now(); + const samples = await client.sendCommand([ + "ZRANGEBYSCORE", + key("worker", workerId, "first_token_samples"), + String(now - windowMs), + String(now) + ]).catch(() => []); + const values = samples.map(parseFirstTokenSample).filter((value) => value > 0); + return { + count: values.length, + p50Ms: percentile(values, 50), + p95Ms: percentile(values, 95) + }; + }; const scoreWorker = async (client, target) => { const workerId = workerIdForTarget(target); const values = await client.mGet([ @@ -8892,10 +8918,20 @@ function createRuntimeRouter({ const workerId = workerIdForTarget(target); const sample = Math.max(0, Math.round(Number(latencyMs) || 0)); const ewmaKey = key("worker", workerId, "ewma_first_token_ms"); + const samplesKey = key("worker", workerId, "first_token_samples"); const prev = readNumber(await client.get(ewmaKey).catch(() => null)); const next = prev > 0 ? Math.round(prev * 0.8 + sample * 0.2) : sample; - const now = String(Date.now()); + const nowMs3 = Date.now(); + const now = String(nowMs3); await client.multi().set(ewmaKey, String(next)).set(key("worker", workerId, "last_first_token_ms"), String(sample)).set(key("worker", workerId, "last_first_token_at"), now).incr(key("worker", workerId, "first_token_count")).set(key("worker", workerId, "heartbeat"), now, { EX: 30 }).exec().catch(() => null); + await client.sendCommand([ + "ZADD", + samplesKey, + String(nowMs3), + `${now}:${sample}:${Math.random().toString(36).slice(2)}` + ]).catch(() => null); + await client.sendCommand(["ZREMRANGEBYSCORE", samplesKey, "0", String(nowMs3 - 60 * 60 * 1e3)]).catch(() => null); + await client.sendCommand(["EXPIRE", samplesKey, String(2 * 60 * 60)]).catch(() => null); }, async getStatus() { const client = await getClient(); @@ -8925,6 +8961,8 @@ function createRuntimeRouter({ key("worker", workerId, "last_first_token_at"), key("worker", workerId, "first_token_count") ]).catch(() => []) : []; + const firstToken5m = await firstTokenWindowStats(client, workerId, 5 * 60 * 1e3); + const firstToken1h = await firstTokenWindowStats(client, workerId, 60 * 60 * 1e3); workers.push({ id: workerId, target, @@ -8949,6 +8987,8 @@ function createRuntimeRouter({ lastFirstTokenMs: readNumber(values?.[18]), lastFirstTokenAt: values?.[19] ? Number(values[19]) : null, firstTokenCount: readNumber(values?.[20]), + firstToken5m, + firstToken1h, score: workerScoreFromValues(values) }); } diff --git a/scripts/runtime-slo-report.mjs b/scripts/runtime-slo-report.mjs index dd52c08..88ec96f 100755 --- a/scripts/runtime-slo-report.mjs +++ b/scripts/runtime-slo-report.mjs @@ -160,6 +160,8 @@ function summarizeRuntime(runtimeJson) { lastFirstTokenMs: worker.lastFirstTokenMs, lastFirstTokenAt: worker.lastFirstTokenAt, firstTokenCount: worker.firstTokenCount, + firstToken5m: worker.firstToken5m ?? null, + firstToken1h: worker.firstToken1h ?? null, cpuLoad: worker.cpuLoad, memoryPressure: worker.memoryPressure, fdPressure: worker.fdPressure, diff --git a/tkmind-proxy.mjs b/tkmind-proxy.mjs index 8ca0ca1..2cabf0a 100644 --- a/tkmind-proxy.mjs +++ b/tkmind-proxy.mjs @@ -124,6 +124,37 @@ function createRuntimeRouter({ const n = Number(value ?? 0); return Number.isFinite(n) ? n : 0; }; + const parseFirstTokenSample = (value) => { + const parts = String(value ?? '').split(':'); + return readNumber(parts[1]); + }; + const percentile = (values, pct) => { + const sorted = values + .map((value) => readNumber(value)) + .filter((value) => value > 0) + .sort((a, b) => a - b); + if (sorted.length === 0) return 0; + const idx = Math.min(sorted.length - 1, Math.max(0, Math.ceil((pct / 100) * sorted.length) - 1)); + return sorted[idx]; + }; + const firstTokenWindowStats = async (client, workerId, windowMs) => { + if (!client) return { count: 0, p50Ms: 0, p95Ms: 0 }; + const now = Date.now(); + const samples = await client + .sendCommand([ + 'ZRANGEBYSCORE', + key('worker', workerId, 'first_token_samples'), + String(now - windowMs), + String(now), + ]) + .catch(() => []); + const values = samples.map(parseFirstTokenSample).filter((value) => value > 0); + return { + count: values.length, + p50Ms: percentile(values, 50), + p95Ms: percentile(values, 95), + }; + }; const scoreWorker = async (client, target) => { const workerId = workerIdForTarget(target); const values = await client.mGet([ @@ -225,9 +256,11 @@ function createRuntimeRouter({ const workerId = workerIdForTarget(target); const sample = Math.max(0, Math.round(Number(latencyMs) || 0)); const ewmaKey = key('worker', workerId, 'ewma_first_token_ms'); + const samplesKey = key('worker', workerId, 'first_token_samples'); const prev = readNumber(await client.get(ewmaKey).catch(() => null)); const next = prev > 0 ? Math.round(prev * 0.8 + sample * 0.2) : sample; - const now = String(Date.now()); + const nowMs = Date.now(); + const now = String(nowMs); await client .multi() .set(ewmaKey, String(next)) @@ -237,6 +270,18 @@ function createRuntimeRouter({ .set(key('worker', workerId, 'heartbeat'), now, { EX: 30 }) .exec() .catch(() => null); + await client + .sendCommand([ + 'ZADD', + samplesKey, + String(nowMs), + `${now}:${sample}:${Math.random().toString(36).slice(2)}`, + ]) + .catch(() => null); + await client + .sendCommand(['ZREMRANGEBYSCORE', samplesKey, '0', String(nowMs - 60 * 60 * 1000)]) + .catch(() => null); + await client.sendCommand(['EXPIRE', samplesKey, String(2 * 60 * 60)]).catch(() => null); }, async getStatus() { const client = await getClient(); @@ -270,6 +315,8 @@ function createRuntimeRouter({ ]) .catch(() => []) : []; + const firstToken5m = await firstTokenWindowStats(client, workerId, 5 * 60 * 1000); + const firstToken1h = await firstTokenWindowStats(client, workerId, 60 * 60 * 1000); workers.push({ id: workerId, target, @@ -294,6 +341,8 @@ function createRuntimeRouter({ lastFirstTokenMs: readNumber(values?.[18]), lastFirstTokenAt: values?.[19] ? Number(values[19]) : null, firstTokenCount: readNumber(values?.[20]), + firstToken5m, + firstToken1h, score: workerScoreFromValues(values), }); }