tkmind/memind
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat: record first token latency in runtime router

Browse Source
This commit is contained in:
John
2026-07-02 07:45:47 +08:00
parent 41b7b5ffa3
commit 8e5094fcd9
4 changed files with 86 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.runtime/portal/scripts/runtime-slo-report.mjs
+5
View File
@@ -155,6 +155,10 @@ function summarizeRuntime(runtimeJson) {
streamOpenCount: worker.streamOpenCount,
streamAbortCount: worker.streamAbortCount,
streamErrorCount: worker.streamErrorCount,
ewmaFirstTokenMs: worker.ewmaFirstTokenMs,
lastFirstTokenMs: worker.lastFirstTokenMs,
lastFirstTokenAt: worker.lastFirstTokenAt,
firstTokenCount: worker.firstTokenCount,
cpuLoad: worker.cpuLoad,
memoryPressure: worker.memoryPressure,
fdPressure: worker.fdPressure,
@@ -213,6 +217,7 @@ for (const worker of runtimeSummary?.workers ?? []) {
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}`);
if (worker.firstTokenCount > 0 && worker.ewmaFirstTokenMs <= 0) failures.push(`${worker.id}_first_token_ewma_missing`);
}
if (runtimeSummary?.toolRuntime?.chatInjectsCodeTools !== false) failures.push('chat_injects_code_tools');
.runtime/portal/server.mjs
+34 -2
View File
@@ -8768,6 +8768,17 @@ function createRuntimeRouter({
if (streamKey) multi.set(streamKey, status, { EX: 600 });
await multi.exec().catch(() => null);
},
async firstTokenObserved(target, latencyMs) {
const client = await getClient();
if (!client || !target) return;
const workerId = workerIdForTarget(target);
const sample = Math.max(0, Math.round(Number(latencyMs) || 0));
const ewmaKey = key("worker", workerId, "ewma_first_token_ms");
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());
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);
},
async getStatus() {
const client = await getClient();
const workers = [];
@@ -8791,7 +8802,10 @@ function createRuntimeRouter({
key("worker", workerId, "fd_count"),
key("worker", workerId, "container_pids"),
key("worker", workerId, "container_health"),
key("worker", workerId, "metrics_sampled_at")
key("worker", workerId, "metrics_sampled_at"),
key("worker", workerId, "last_first_token_ms"),
key("worker", workerId, "last_first_token_at"),
key("worker", workerId, "first_token_count")
]).catch(() => []) : [];
workers.push({
id: workerId,
@@ -8814,6 +8828,9 @@ function createRuntimeRouter({
containerPids: readNumber(values?.[15]),
containerHealth: values?.[16] ?? null,
metricsSampledAt: values?.[17] ? Number(values[17]) : null,
lastFirstTokenMs: readNumber(values?.[18]),
lastFirstTokenAt: values?.[19] ? Number(values[19]) : null,
firstTokenCount: readNumber(values?.[20]),
score: workerScoreFromValues(values)
});
}
@@ -9344,6 +9361,10 @@ function createTkmindProxy({
if (!runtimeRouter || !sessionId || !target) return;
await runtimeRouter.streamEnded(sessionId, target, options).catch(() => null);
}
function markFirstTokenObserved(target, latencyMs) {
if (!runtimeRouter || !target) return;
void runtimeRouter.firstTokenObserved(target, latencyMs).catch(() => null);
}
async function getRuntimeStatus() {
const targetStatuses = [];
for (const target of targets) {
@@ -9859,6 +9880,7 @@ function createTkmindProxy({
];
const proxySessionEvents = async (req, res, sessionId, { onAfterFinish, onEvent } = {}) => {
const upstreamAbort = new AbortController();
const streamRequestedAt = Date.now();
let clientClosed = false;
const abortUpstream = () => {
clientClosed = true;
@@ -9916,6 +9938,16 @@ function createTkmindProxy({
}
});
const source = Readable.fromWeb(upstream.body);
let firstChunkSeen = false;
const firstTokenProbe = new Transform2({
transform(chunk, _encoding, callback) {
if (!firstChunkSeen && chunk?.length > 0) {
firstChunkSeen = true;
markFirstTokenObserved(sessionTarget, Date.now() - streamRequestedAt);
}
callback(null, chunk);
}
});
const linkSanitizer = createSessionEventSanitizer(req.currentUser, { onEvent });
const waitForDrain = () => new Promise((resolve) => res.once("drain", resolve));
const writeClientChunk = async (chunk) => {
@@ -9945,7 +9977,7 @@ function createTkmindProxy({
let streamCloseStatus = "closed";
try {
try {
await pipeline(source, linkSanitizer, billingTransform, clientSink);
await pipeline(source, firstTokenProbe, linkSanitizer, billingTransform, clientSink);
} catch (err) {
streamCloseStatus = clientClosed || upstreamAbort.signal.aborted ? "aborted" : "error";
throw err;
scripts/runtime-slo-report.mjs
+5
View File
@@ -155,6 +155,10 @@ function summarizeRuntime(runtimeJson) {
streamOpenCount: worker.streamOpenCount,
streamAbortCount: worker.streamAbortCount,
streamErrorCount: worker.streamErrorCount,
ewmaFirstTokenMs: worker.ewmaFirstTokenMs,
lastFirstTokenMs: worker.lastFirstTokenMs,
lastFirstTokenAt: worker.lastFirstTokenAt,
firstTokenCount: worker.firstTokenCount,
cpuLoad: worker.cpuLoad,
memoryPressure: worker.memoryPressure,
fdPressure: worker.fdPressure,
@@ -213,6 +217,7 @@ for (const worker of runtimeSummary?.workers ?? []) {
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}`);
if (worker.firstTokenCount > 0 && worker.ewmaFirstTokenMs <= 0) failures.push(`${worker.id}_first_token_ewma_missing`);
}
if (runtimeSummary?.toolRuntime?.chatInjectsCodeTools !== false) failures.push('chat_injects_code_tools');
tkmind-proxy.mjs
+42 -1
View File
@@ -219,6 +219,25 @@ function createRuntimeRouter({
if (streamKey) multi.set(streamKey, status, { EX: 600 });
await multi.exec().catch(() => null);
},
async firstTokenObserved(target, latencyMs) {
const client = await getClient();
if (!client || !target) return;
const workerId = workerIdForTarget(target);
const sample = Math.max(0, Math.round(Number(latencyMs) || 0));
const ewmaKey = key('worker', workerId, 'ewma_first_token_ms');
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());
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);
},
async getStatus() {
const client = await getClient();
const workers = [];
@@ -245,6 +264,9 @@ function createRuntimeRouter({
key('worker', workerId, 'container_pids'),
key('worker', workerId, 'container_health'),
key('worker', workerId, 'metrics_sampled_at'),
key('worker', workerId, 'last_first_token_ms'),
key('worker', workerId, 'last_first_token_at'),
key('worker', workerId, 'first_token_count'),
])
.catch(() => [])
: [];
@@ -269,6 +291,9 @@ function createRuntimeRouter({
containerPids: readNumber(values?.[15]),
containerHealth: values?.[16] ?? null,
metricsSampledAt: values?.[17] ? Number(values[17]) : null,
lastFirstTokenMs: readNumber(values?.[18]),
lastFirstTokenAt: values?.[19] ? Number(values[19]) : null,
firstTokenCount: readNumber(values?.[20]),
score: workerScoreFromValues(values),
});
}
@@ -909,6 +934,11 @@ export function createTkmindProxy({
await runtimeRouter.streamEnded(sessionId, target, options).catch(() => null);
}
function markFirstTokenObserved(target, latencyMs) {
if (!runtimeRouter || !target) return;
void runtimeRouter.firstTokenObserved(target, latencyMs).catch(() => null);
}
async function getRuntimeStatus() {
const targetStatuses = [];
for (const target of targets) {
@@ -1503,6 +1533,7 @@ export function createTkmindProxy({
const proxySessionEvents = async (req, res, sessionId, { onAfterFinish, onEvent } = {}) => {
const upstreamAbort = new AbortController();
const streamRequestedAt = Date.now();
let clientClosed = false;
const abortUpstream = () => {
clientClosed = true;
@@ -1564,6 +1595,16 @@ export function createTkmindProxy({
});
const source = Readable.fromWeb(upstream.body);
let firstChunkSeen = false;
const firstTokenProbe = new Transform({
transform(chunk, _encoding, callback) {
if (!firstChunkSeen && chunk?.length > 0) {
firstChunkSeen = true;
markFirstTokenObserved(sessionTarget, Date.now() - streamRequestedAt);
}
callback(null, chunk);
},
});
const linkSanitizer = createSessionEventSanitizer(req.currentUser, { onEvent });
const waitForDrain = () => new Promise((resolve) => res.once('drain', resolve));
const writeClientChunk = async (chunk) => {
@@ -1595,7 +1636,7 @@ export function createTkmindProxy({
let streamCloseStatus = 'closed';
try {
try {
await pipeline(source, linkSanitizer, billingTransform, clientSink);
await pipeline(source, firstTokenProbe, linkSanitizer, billingTransform, clientSink);
} catch (err) {
streamCloseStatus = clientClosed || upstreamAbort.signal.aborted ? 'aborted' : 'error';
throw err;