From e10fd7271559d14749672bfbf7e091bc1ff77f8e Mon Sep 17 00:00:00 2001 From: Leonardo Cardoso Date: Fri, 19 Jun 2026 22:49:01 +0200 Subject: [PATCH 1/2] feat: telemetry endpoints (/status extras, /metrics, /usage) Expose the router-only telemetry no single backend can produce: queue backlog, swap economics, rolling throughput, cumulative counters, a recent-request log, and error/health signal. Jano already sees every request/response and invokes every swap, so this is mostly counters and small ring buffers in existing code paths. - extend GET /status with inFlight, oldestWaitingMs, currentModelLoadedAt, lastSwapDurationMs, swaps{15m,Hour}, recent{Gen,Prompt}TokS, tokSByModel, uptime, cumulative counters, requestsPerMinute, errors, backendHealth - add GET /metrics (Prometheus: counters, queue gauges, swap-duration histogram + per-transition cost table, backend health) - add GET /usage?limit=N (recent per-request token/timing records) - telemetry.ts: bounded ring buffers + counters, clock-injectable - responseStats.ts: normalize OpenAI usage / llama.cpp timings / Ollama nanosecond fields to tok/s; bodyTap.ts captures stats without buffering - USAGE_RECORDS_MAX env (default 500) Hardware metrics and relay heartbeat stay out of scope (sidecar's job). 35 new tests. --- .env.example | 5 + README.md | 93 ++++++++- src/bodyTap.ts | 60 ++++++ src/config.ts | 5 + src/dispatcher.ts | 38 +++- src/index.ts | 101 +++++++++- src/metrics.ts | 13 ++ src/queue.ts | 10 + src/responseStats.ts | 125 ++++++++++++ src/telemetry.ts | 366 ++++++++++++++++++++++++++++++++++++ tests/bodyTap.test.ts | 36 ++++ tests/responseStats.test.ts | 145 ++++++++++++++ tests/telemetry-e2e.test.ts | 258 +++++++++++++++++++++++++ tests/telemetry.test.ts | 201 ++++++++++++++++++++ 14 files changed, 1450 insertions(+), 6 deletions(-) create mode 100644 src/bodyTap.ts create mode 100644 src/metrics.ts create mode 100644 src/responseStats.ts create mode 100644 src/telemetry.ts create mode 100644 tests/bodyTap.test.ts create mode 100644 tests/responseStats.test.ts create mode 100644 tests/telemetry-e2e.test.ts create mode 100644 tests/telemetry.test.ts diff --git a/.env.example b/.env.example index 997be1e..073c9de 100644 --- a/.env.example +++ b/.env.example @@ -32,3 +32,8 @@ HEALTH_POLL_INTERVAL_MS=1000 # How long upstream chat-completions calls may take. Generous default for # slow CPU / cold-cache generations on local models. REQUEST_TIMEOUT_MS=600000 + +# How many recent request records jano keeps in memory for `GET /usage` +# (the "what did my last N requests cost" log). Older records drop FIFO. +# Purely observability; bump it for a longer benchmark history. +USAGE_RECORDS_MAX=500 diff --git a/README.md b/README.md index 8dc8cff..45b1672 100644 --- a/README.md +++ b/README.md @@ -150,17 +150,103 @@ Returns the configured models in OpenAI's list format. Useful for clients that p ### `GET /health` and `GET /status` -Both return the same payload, jano's view of the world: +Both return the same payload: jano's full view of the world. Because jano sits +_above_ the backends and owns every request, response, and swap, it can report +telemetry no single backend (Ollama, `llama-server`, `mlx-lm.server`) can +produce on its own — queue backlog, swap economics, rolling throughput across +all callers, and cumulative counters. -```json +```jsonc { "ok": true, "currentModel": "chat", + "currentModelLoadedAt": "2026-06-19T18:02:11.000Z", + + // Queue & concurrency (router-only). "queueDepth": 0, - "queueByModel": { "chat": 0, "code": 0, "fast": 0 } + "queueByModel": { "chat": 0, "code": 0 }, + "inFlight": 1, + "oldestWaitingMs": 0, // age of the head-of-line waiting request + + // Swap economics (router-only — only jano invokes the swap). + "lastSwapDurationMs": 21840, + "swapsLast15m": 1, + "swapsLastHour": 3, + + // Aggregate "lately" throughput across ALL callers (rolling, last N). + "recentGenTokS": 47.2, + "recentPromptTokS": 312.5, + "tokSByModel": { "chat": 51.0, "code": 39.4 }, + + // Cumulative counters since start. + "uptimeSeconds": 84211, + "requestsServedTotal": 128, + "requestsByModel": { "chat": 90, "code": 38 }, + "tokensGeneratedTotal": 51234, + "tokensByModel": { "chat": 33001, "code": 18233 }, + "requestsPerMinute": 4, + + // Error & health signal. + "errorsLast15m": 0, + "lastError": null, + "backendHealth": { "chat": "up", "code": "up" }, } ``` +Token counts and tok/s come from the backend's response when it reports them — +jano normalizes across shapes: OpenAI `usage`, llama.cpp `timings`, and +Ollama-style nanosecond fields (`eval_count`/`eval_duration`). Generation tok/s +is otherwise derived from measured time-to-first-byte and completion. When a +backend reports nothing usable, the relevant fields are simply `null` — +telemetry degrades gracefully rather than guessing. + +### `GET /metrics` + +Prometheus text exposition (`v0.0.4`) of the cumulative counters, swap-duration +histogram, queue gauges, and per-backend health — scrape it straight into +Grafana, or read it from a client. Sample: + +```text +jano_uptime_seconds 84211 +jano_in_flight 1 +jano_queue_depth{model="chat"} 0 +jano_requests_served_total{model="chat"} 90 +jano_tokens_generated_total{model="chat"} 33001 +jano_swaps_total{from="chat",to="code"} 3 +jano_swap_transition_duration_milliseconds_sum{from="chat",to="code"} 65520 +jano_swap_duration_milliseconds_bucket{le="30000"} 3 +jano_backend_up{model="chat"} 1 +``` + +### `GET /usage?limit=50` + +A ring buffer of the most recent requests (newest first), each with token +counts and timings — "what did my last N requests cost in compute." The buffer +size is `USAGE_RECORDS_MAX` (default 500). + +```jsonc +{ + "count": 2, + "records": [ + { + "ts": "2026-06-19T18:05:42.000Z", + "model": "chat", + "status": 200, + "total_ms": 1840, + "ttfb_ms": 120, + "prompt_tokens": 312, + "gen_tokens": 87, + "gen_tok_s": 51.0, + "prompt_tok_s": 410.2, + }, + ], +} +``` + +> **Out of scope:** GPU temperature, utilization, fan speed, and system +> RAM/CPU/power are hardware readings outside any router's reach. Jano does +> not and cannot report them — that's a host-side helper's job. + ## The swap script contract Jano shells out to `SWAP_COMMAND` to swap models. The contract is intentionally small: @@ -205,6 +291,7 @@ All via env vars; copy `.env.example` to `.env` to start. | `SWAP_WAIT_TIMEOUT_MS` | no | `180000` | Max time jano waits for a backend to become healthy after a swap. | | `HEALTH_POLL_INTERVAL_MS` | no | `1000` | How often jano pings `/health` while waiting. | | `REQUEST_TIMEOUT_MS` | no | `600000` | Per-request upstream timeout. Generous default for cold/large generations. | +| `USAGE_RECORDS_MAX` | no | `500` | Size of the in-memory recent-request ring served by `GET /usage`. | ## When you don't need jano diff --git a/src/bodyTap.ts b/src/bodyTap.ts new file mode 100644 index 0000000..3960fd4 --- /dev/null +++ b/src/bodyTap.ts @@ -0,0 +1,60 @@ +import { Buffer } from 'node:buffer'; + +/** + * Captures just enough of a forwarded response body to extract token/timing + * telemetry, without buffering the whole thing or perturbing what we stream + * to the client. + * + * Two modes, chosen by `tailOnly`: + * + * - **Streaming (`tailOnly = true`)**: keep only the last `cap` bytes. The + * `usage`/`timings` payload an SSE response carries always lives in the + * final `data:` frame, so a small tail window captures it while bounding + * memory regardless of how long the generation runs. + * - **Non-streaming (`tailOnly = false`)**: a chat-completions JSON body is + * small, so accumulate the whole thing — but stop at `cap` to stay safe + * against a pathological upstream. If we overflow we simply decline to + * parse (telemetry records null token counts rather than risk garbage). + * + * Feeding chunks is allocation-light: we hold references and only concat once, + * lazily, in `text()`. + */ +export class BodyTap { + private chunks: Buffer[] = []; + private bytes = 0; + private overflowed = false; + private readonly tailOnly: boolean; + private readonly cap: number; + + constructor(tailOnly: boolean, cap = 64 * 1024) { + this.tailOnly = tailOnly; + this.cap = cap; + } + + push(chunk: Buffer): void { + if (this.tailOnly) { + this.chunks.push(chunk); + this.bytes += chunk.length; + // Drop from the front until we're back under cap, but always keep at + // least the most recent chunk so a single oversized frame still parses. + while (this.bytes > this.cap && this.chunks.length > 1) { + const removed = this.chunks.shift(); + if (removed) this.bytes -= removed.length; + } + return; + } + if (this.overflowed) return; + if (this.bytes + chunk.length > this.cap) { + this.overflowed = true; + return; + } + this.chunks.push(chunk); + this.bytes += chunk.length; + } + + /** Decoded captured bytes, or null if a non-streaming body overflowed. */ + text(): string | null { + if (this.overflowed) return null; + return Buffer.concat(this.chunks).toString('utf8'); + } +} diff --git a/src/config.ts b/src/config.ts index dd67087..8c8d075 100644 --- a/src/config.ts +++ b/src/config.ts @@ -135,6 +135,11 @@ export const env = { /** Reset a model's swap-failure count after it has been quiet this long; * lets the dispatcher retry once a broken backend may have recovered. */ SWAP_FAILURE_RESET_MS: optInt('SWAP_FAILURE_RESET_MS', 5 * 60_000), + + /** Size of the in-memory recent-request ring served by `GET /usage`. Older + * records are dropped FIFO. Purely an observability buffer; bump it if you + * want a longer benchmark history at the cost of a little memory. */ + USAGE_RECORDS_MAX: optInt('USAGE_RECORDS_MAX', 500), }; export const models: ModelDef[] = loadModels(env.MODELS_FILE); diff --git a/src/dispatcher.ts b/src/dispatcher.ts index e07121d..5e6862a 100644 --- a/src/dispatcher.ts +++ b/src/dispatcher.ts @@ -1,8 +1,9 @@ import type { ModelName } from './types.ts'; import { env, modelNames } from './config.ts'; import { FailureBudget } from './failureBudget.ts'; -import { _internal, getQueueByModel, getQueueDepth } from './queue.ts'; +import { _internal, getOldestEnqueuedAt, getQueueByModel, getQueueDepth } from './queue.ts'; import { detectLoaded, ensureLoaded } from './swap.ts'; +import { telemetry } from './metrics.ts'; import { log } from './log.ts'; let currentModel: ModelName | null = null; @@ -15,11 +16,29 @@ let running = false; // Reproduces the 2026-05-12 wedge fix. const swapFailures = new FailureBudget(env.SWAP_FAILURE_LIMIT, env.SWAP_FAILURE_RESET_MS); +/** A request that never reached a backend (dead client, swap failure) still + * belongs in the usage history with its short-circuit status and no tokens. */ +function recordShortCircuit(model: ModelName, status: number, enqueuedAt: number): void { + telemetry.recordRequest({ + model, + status, + totalMs: Date.now() - enqueuedAt, + ttfbMs: null, + promptTokens: null, + genTokens: null, + genTokS: null, + promptTokS: null, + }); +} + export function getStatus() { + const oldest = getOldestEnqueuedAt(); return { currentModel, queueDepth: getQueueDepth(), queueByModel: getQueueByModel(modelNames()), + oldestWaitingMs: oldest === null ? 0 : Math.max(0, Date.now() - oldest), + ...telemetry.snapshot(), }; } @@ -35,6 +54,7 @@ async function loop(): Promise { if (!task.isAlive()) { log.info('dropping disconnected task', { id: task.id, model: task.model }); + recordShortCircuit(task.model, 499, task.enqueuedAt); await task.fail(499, 'client disconnected before dispatch'); continue; } @@ -47,14 +67,18 @@ async function loop(): Promise { model: task.model, fails, }); + telemetry.recordError(task.model, `fail-fast: ${fails} consecutive swap failures`); + recordShortCircuit(task.model, 503, task.enqueuedAt); await task.fail( 503, `backend "${task.model}" is unavailable (${fails} consecutive swap failures; auto-retry in ~${Math.round(env.SWAP_FAILURE_RESET_MS / 1000)}s)` ); continue; } + const swapStart = Date.now(); try { await ensureLoaded(task.model); + telemetry.recordSwap(currentModel, task.model, Date.now() - swapStart); currentModel = task.model; swapFailures.recordSuccess(task.model); } catch (err) { @@ -65,16 +89,24 @@ async function loop(): Promise { fails: newFails, err: String(err), }); + telemetry.recordError(task.model, `swap failed: ${String(err)}`); + recordShortCircuit(task.model, 503, task.enqueuedAt); await task.fail(503, `swap to "${task.model}" failed: ${String(err)}`); continue; } } try { - await task.run(); + telemetry.incInFlight(); + try { + await task.run(); + } finally { + telemetry.decInFlight(); + } swapFailures.recordSuccess(task.model); } catch (err) { log.error('task threw', { id: task.id, err: String(err) }); + telemetry.recordError(task.model, `task threw: ${String(err)}`); } } log.info('dispatcher stopped'); @@ -83,7 +115,9 @@ async function loop(): Promise { export async function start(): Promise { if (running) return; running = true; + telemetry.seedModels(modelNames()); currentModel = await detectLoaded(); + if (currentModel !== null) telemetry.noteInitialModel(currentModel); log.info('dispatcher detected initial model', { currentModel }); void loop(); } diff --git a/src/index.ts b/src/index.ts index ca93b04..4c9750d 100644 --- a/src/index.ts +++ b/src/index.ts @@ -4,7 +4,10 @@ import { request as undiciRequest } from 'undici'; import { backendUrl, env, models } from './config.ts'; import { getStatus, start, stop } from './dispatcher.ts'; import { log } from './log.ts'; +import { telemetry } from './metrics.ts'; import { pickModel } from './picker.ts'; +import { BodyTap } from './bodyTap.ts'; +import { parseResponseStats } from './responseStats.ts'; import { enqueue, type Task } from './queue.ts'; import type { ModelDef, ModelName } from './types.ts'; @@ -29,6 +32,43 @@ function sendJson(res: http.ServerResponse, status: number, body: unknown): void res.end(JSON.stringify(body)); } +/** + * Turn a finished chat-completions request into a telemetry record. Pulls + * token counts + tok/s from the tapped response body when the backend reports + * them (llama.cpp `timings` / OpenAI `usage`); otherwise derives generation + * tok/s from the measured time between first byte and completion. + */ +function recordCompletion(o: { + model: ModelName; + status: number; + enqueuedAt: number; + runStart: number; + ttfbMs: number | null; + doneMs: number; + tapText: string | null; + isStream: boolean; +}): void { + const stats = o.tapText === null ? null : parseResponseStats(o.tapText, o.isStream); + const promptTokens = stats?.promptTokens ?? null; + const genTokens = stats?.genTokens ?? null; + const promptTokS = stats?.promptTokS ?? null; + let genTokS = stats?.genTokS ?? null; + if (genTokS === null && genTokens !== null && o.ttfbMs !== null) { + const genMs = o.doneMs - (o.runStart + o.ttfbMs); + if (genMs > 0) genTokS = genTokens / (genMs / 1000); + } + telemetry.recordRequest({ + model: o.model, + status: o.status, + totalMs: o.doneMs - o.enqueuedAt, + ttfbMs: o.ttfbMs, + promptTokens, + genTokens, + genTokS, + promptTokS, + }); +} + // Hop-by-hop headers per RFC 7230 §6.1; never forward these. const HOP_BY_HOP = new Set([ 'connection', @@ -204,6 +244,7 @@ async function handleChatCompletions( return Promise.resolve(); }, run: async () => { + const runStart = Date.now(); const upstream = `${backendUrl(model)}/v1/chat/completions`; let upstreamRes: Awaited>; try { @@ -225,11 +266,20 @@ async function handleChatCompletions( if (!res.destroyed) { res.writeHead(upstreamRes.statusCode, copyHeaders(upstreamRes.headers)); } + // Tap the body for token/timing telemetry without buffering the whole + // stream: a tail window for SSE, the small full body for one-shot JSON. + const ctRaw = upstreamRes.headers['content-type']; + const contentType = Array.isArray(ctRaw) ? ctRaw.join(' ') : (ctRaw ?? ''); + const isStream = contentType.includes('text/event-stream'); + const tap = new BodyTap(isStream); + let ttfbMs: number | null = null; // Stream regardless of stream:true: undici gives us an async iterator // either way, and pumping it is correct for both SSE and one-shot JSON. try { for await (const chunk of upstreamRes.body) { if (clientDisconnected || res.destroyed) break; + if (ttfbMs === null) ttfbMs = Date.now() - runStart; + tap.push(chunk as Buffer); if (!res.write(chunk)) { await new Promise((resolve) => { const done = () => { @@ -251,13 +301,26 @@ async function handleChatCompletions( throw err; } if (!res.writableEnded) res.end(); + const doneMs = Date.now(); log.info('served', { id, model, status: upstreamRes.statusCode, - ms: Date.now() - task.enqueuedAt, + ms: doneMs - task.enqueuedAt, aborted: clientDisconnected, }); + recordCompletion({ + model, + // A disconnect-truncated stream is not a clean serve: record it as 499 + // with null tokens so it doesn't inflate served-total or throughput. + status: clientDisconnected ? 499 : upstreamRes.statusCode, + enqueuedAt: task.enqueuedAt, + runStart, + ttfbMs, + doneMs, + tapText: clientDisconnected ? null : tap.text(), + isStream, + }); }, }; @@ -276,6 +339,16 @@ async function handleChatCompletions( // Socket may already be gone. } } + telemetry.recordRequest({ + model, + status: 502, + totalMs: Date.now() - task.enqueuedAt, + ttfbMs: null, + promptTokens: null, + genTokens: null, + genTokS: null, + promptTokS: null, + }); throw err; } }; @@ -352,6 +425,24 @@ function handleStatus(_req: http.IncomingMessage, res: http.ServerResponse): voi sendJson(res, 200, { ok: true, ...getStatus() }); } +function handleUsage(req: http.IncomingMessage, res: http.ServerResponse): void { + const url = new URL(req.url ?? '/', `http://${req.headers.host ?? 'localhost'}`); + const rawLimit = Number(url.searchParams.get('limit') ?? '50'); + const limit = Number.isFinite(rawLimit) && rawLimit > 0 ? Math.floor(rawLimit) : 50; + const records = telemetry.usageList(limit); + sendJson(res, 200, { count: records.length, records }); +} + +function handleMetrics(_req: http.IncomingMessage, res: http.ServerResponse): void { + const status = getStatus(); + const body = telemetry.prometheus({ + queueByModel: status.queueByModel, + currentModel: status.currentModel, + }); + res.writeHead(200, { 'Content-Type': 'text/plain; version=0.0.4; charset=utf-8' }); + res.end(body); +} + const server = http.createServer((req, res) => { const url = new URL(req.url ?? '/', `http://${req.headers.host ?? 'localhost'}`); const route = `${req.method} ${url.pathname}`; @@ -371,6 +462,14 @@ const server = http.createServer((req, res) => { handleStatus(req, res); return; } + if (route === 'GET /usage') { + handleUsage(req, res); + return; + } + if (route === 'GET /metrics') { + handleMetrics(req, res); + return; + } sendJson(res, 404, { error: `no route for ${route}` }); }); diff --git a/src/metrics.ts b/src/metrics.ts new file mode 100644 index 0000000..6937dbf --- /dev/null +++ b/src/metrics.ts @@ -0,0 +1,13 @@ +import { env } from './config.ts'; +import { Telemetry } from './telemetry.ts'; + +/** + * The live, process-wide telemetry instance shared by the dispatcher (which + * records swaps, errors, and in-flight state) and the HTTP layer (which + * records per-request token/timing data and serves `/status`, `/usage`, + * `/metrics`). + * + * Kept in its own module so `telemetry.ts` stays env-free and unit-testable — + * importing the class there must not drag in config loading. + */ +export const telemetry = new Telemetry({ usageMax: env.USAGE_RECORDS_MAX }); diff --git a/src/queue.ts b/src/queue.ts index d7c9195..024f86b 100644 --- a/src/queue.ts +++ b/src/queue.ts @@ -29,6 +29,16 @@ export function getQueueByModel(modelNames: ModelName[]): Record return counts; } +/** Enqueue time of the oldest waiting task, or null if the queue is empty. + * Powers the `oldestWaitingMs` "am I backed up?" signal on `/status`. */ +export function getOldestEnqueuedAt(): number | null { + let oldest: number | null = null; + for (const t of queue) { + if (oldest === null || t.enqueuedAt < oldest) oldest = t.enqueuedAt; + } + return oldest; +} + export function enqueue(task: Task): void { queue.push(task); wakeUp?.(); diff --git a/src/responseStats.ts b/src/responseStats.ts new file mode 100644 index 0000000..cf25625 --- /dev/null +++ b/src/responseStats.ts @@ -0,0 +1,125 @@ +/** + * Pure helpers for pulling token counts and timings out of a backend's + * response, without disturbing the bytes we forward to the client. + * + * Jano forwards request bodies untouched and streams responses through, so + * the only place token-level telemetry can come from is the response itself. + * Two shapes show up in practice: + * + * - OpenAI-style JSON (`usage: { prompt_tokens, completion_tokens }`), + * present in non-streaming completions and — only when the caller asks + * for `stream_options.include_usage` — in the final SSE chunk. + * - llama.cpp's `timings` block (`predicted_per_second`, `predicted_n`, + * `prompt_per_second`, `prompt_n`), which carries tok/s directly so we + * don't have to derive it. + * + * When neither is present (the common streaming case) we return null and the + * caller falls back to deriving tok/s from wall-clock timing, or records the + * request with null token counts. Telemetry degrades gracefully either way. + */ + +export type ResponseStats = { + promptTokens: number | null; + genTokens: number | null; + /** Generation tok/s as reported by the backend (llama.cpp `timings`). */ + genTokS: number | null; + /** Prompt-processing tok/s as reported by the backend. */ + promptTokS: number | null; +}; + +function num(v: unknown): number | null { + return typeof v === 'number' && Number.isFinite(v) ? v : null; +} + +function isRecord(v: unknown): v is Record { + return typeof v === 'object' && v !== null; +} + +/** tok/s from a token count and a duration in nanoseconds (Ollama units). */ +function perSecond(count: number | null, durationNs: number | null): number | null { + if (count === null || durationNs === null || durationNs <= 0) return null; + return count / (durationNs / 1e9); +} + +/** Extract whatever stats a single parsed JSON object carries, or null. */ +function fromObject(obj: unknown): ResponseStats | null { + if (!isRecord(obj)) return null; + + let promptTokens: number | null = null; + let genTokens: number | null = null; + let genTokS: number | null = null; + let promptTokS: number | null = null; + + if (isRecord(obj.usage)) { + promptTokens = num(obj.usage.prompt_tokens); + genTokens = num(obj.usage.completion_tokens); + } + if (isRecord(obj.timings)) { + // llama.cpp. Prefer usage counts when both exist; timings fills the gaps + // and is the only source of backend-measured tok/s. + promptTokens = promptTokens ?? num(obj.timings.prompt_n); + genTokens = genTokens ?? num(obj.timings.predicted_n); + genTokS = num(obj.timings.predicted_per_second); + promptTokS = num(obj.timings.prompt_per_second); + } + + // Ollama-style native timing: top-level token counts + durations in + // nanoseconds. Jano normalizes these to tok/s so a backend speaking + // Ollama's shape lights up the same throughput telemetry as llama.cpp. + promptTokens = promptTokens ?? num(obj.prompt_eval_count); + genTokens = genTokens ?? num(obj.eval_count); + if (genTokS === null) genTokS = perSecond(num(obj.eval_count), num(obj.eval_duration)); + if (promptTokS === null) { + promptTokS = perSecond(num(obj.prompt_eval_count), num(obj.prompt_eval_duration)); + } + + if (promptTokens === null && genTokens === null && genTokS === null && promptTokS === null) { + return null; + } + return { promptTokens, genTokens, genTokS, promptTokS }; +} + +/** Later non-null fields win — usage/timings live in the final SSE chunk. */ +function mergePreferLater(a: ResponseStats | null, b: ResponseStats): ResponseStats { + if (!a) return b; + return { + promptTokens: b.promptTokens ?? a.promptTokens, + genTokens: b.genTokens ?? a.genTokens, + genTokS: b.genTokS ?? a.genTokS, + promptTokS: b.promptTokS ?? a.promptTokS, + }; +} + +/** + * Parse stats from a (possibly partial) response body. + * + * - Non-streaming: the whole body is one JSON object. + * - Streaming (SSE): scan `data:` frames and keep the last token/timing data + * we find. `[DONE]` and unparseable frames are skipped. + */ +export function parseResponseStats(text: string, isStream: boolean): ResponseStats | null { + if (!isStream) { + try { + return fromObject(JSON.parse(text)); + } catch { + return null; + } + } + + let acc: ResponseStats | null = null; + for (const rawLine of text.split('\n')) { + const line = rawLine.trim(); + if (!line.startsWith('data:')) continue; + const payload = line.slice('data:'.length).trim(); + if (payload === '' || payload === '[DONE]') continue; + let obj: unknown; + try { + obj = JSON.parse(payload); + } catch { + continue; + } + const s = fromObject(obj); + if (s) acc = mergePreferLater(acc, s); + } + return acc; +} diff --git a/src/telemetry.ts b/src/telemetry.ts new file mode 100644 index 0000000..19dd74d --- /dev/null +++ b/src/telemetry.ts @@ -0,0 +1,366 @@ +import type { ModelName } from './types.ts'; + +/** + * In-process telemetry aggregator. This is Jano's wedge over a plain backend: + * because Jano sees every request, every response, and invokes every swap, it + * can serve queue depth, swap economics, rolling throughput, cumulative + * counters, and a recent-request log that no single backend (Ollama, + * llama-server, mlx-lm.server) can produce on its own. + * + * Everything lives in bounded ring buffers + counters; nothing is persisted. + * + * Pure-ish and clock-injectable (like {@link FailureBudget}) so the + * aggregation logic is unit-testable without spinning up the server. The live + * singleton is created in `metrics.ts`; the dispatcher and HTTP layer share it. + */ + +/** One completed (or failed) request, as surfaced by `GET /usage`. */ +export type RequestRecord = { + /** Completion time, epoch ms. */ + ts: number; + model: ModelName; + status: number; + /** Wall-clock from enqueue to fully handled. */ + totalMs: number; + /** Time to first byte from dispatch, or null if never reached. */ + ttfbMs: number | null; + promptTokens: number | null; + genTokens: number | null; + genTokS: number | null; + promptTokS: number | null; +}; + +export type SwapRecord = { + ts: number; + from: ModelName | null; + to: ModelName; + durationMs: number; +}; + +export type ErrorRecord = { + ts: number; + model: ModelName | null; + message: string; +}; + +export type Health = 'up' | 'down' | 'unknown'; + +const MS_15M = 15 * 60_000; +const MS_1H = 60 * 60_000; +const MS_1M = 60_000; + +/** How many recent completions feed the rolling tok/s figures. */ +const RECENT_SAMPLES = 30; +const SWAP_HISTORY_MAX = 500; +const ERROR_HISTORY_MAX = 200; + +/** Cumulative histogram bucket upper bounds for swap duration, in ms. */ +const SWAP_BUCKETS_MS = [1_000, 5_000, 15_000, 30_000, 60_000, 120_000, 300_000]; + +function round1(n: number): number { + return Math.round(n * 10) / 10; +} + +function mean(xs: number[]): number | null { + if (xs.length === 0) return null; + return xs.reduce((a, b) => a + b, 0) / xs.length; +} + +function isOk(status: number): boolean { + return status >= 200 && status < 300; +} + +export class Telemetry { + private readonly now: () => number; + private readonly usageMax: number; + private readonly startedAt: number; + + private inFlight = 0; + private currentModelLoadedAt: number | null = null; + private lastSwapDurationMs: number | null = null; + + private requests: RequestRecord[] = []; + private swaps: SwapRecord[] = []; + private errors: ErrorRecord[] = []; + + private requestsServedTotal = 0; + private readonly requestsByModel = new Map(); + private tokensGeneratedTotal = 0; + private readonly tokensByModel = new Map(); + private promptTokensTotal = 0; + private errorsTotal = 0; + private swapsTotal = 0; + private readonly swapByTransition = new Map(); + + private readonly health = new Map(); + + constructor(opts?: { now?: () => number; usageMax?: number }) { + this.now = opts?.now ?? Date.now; + this.usageMax = Math.max(1, opts?.usageMax ?? 500); + this.startedAt = this.now(); + } + + // ---- mutation ----------------------------------------------------------- + + /** Register known models so `backendHealth` lists them before any traffic. */ + seedModels(names: ModelName[]): void { + for (const n of names) if (!this.health.has(n)) this.health.set(n, 'unknown'); + } + + incInFlight(): void { + this.inFlight++; + } + + decInFlight(): void { + if (this.inFlight > 0) this.inFlight--; + } + + /** Initial loaded model detected at startup — sets loadedAt, counts no swap. */ + noteInitialModel(model: ModelName): void { + this.currentModelLoadedAt = this.now(); + this.health.set(model, 'up'); + } + + recordSwap(from: ModelName | null, to: ModelName, durationMs: number): void { + const ts = this.now(); + this.swaps.push({ ts, from, to, durationMs }); + if (this.swaps.length > SWAP_HISTORY_MAX) this.swaps.shift(); + this.swapsTotal++; + this.lastSwapDurationMs = durationMs; + this.currentModelLoadedAt = ts; + this.health.set(to, 'up'); + + const key = `${from ?? '(none)'}→${to}`; + const agg = this.swapByTransition.get(key) ?? { count: 0, totalMs: 0 }; + agg.count++; + agg.totalMs += durationMs; + this.swapByTransition.set(key, agg); + } + + recordRequest(rec: Omit): void { + const full: RequestRecord = { ts: this.now(), ...rec }; + this.requests.push(full); + if (this.requests.length > this.usageMax) this.requests.shift(); + + if (isOk(rec.status)) { + this.requestsServedTotal++; + this.requestsByModel.set(rec.model, (this.requestsByModel.get(rec.model) ?? 0) + 1); + this.health.set(rec.model, 'up'); + if (rec.genTokens !== null) { + this.tokensGeneratedTotal += rec.genTokens; + this.tokensByModel.set(rec.model, (this.tokensByModel.get(rec.model) ?? 0) + rec.genTokens); + } + if (rec.promptTokens !== null) this.promptTokensTotal += rec.promptTokens; + } + } + + recordError(model: ModelName | null, message: string): void { + this.errors.push({ ts: this.now(), model, message }); + if (this.errors.length > ERROR_HISTORY_MAX) this.errors.shift(); + this.errorsTotal++; + if (model !== null) this.health.set(model, 'down'); + } + + // ---- aggregation -------------------------------------------------------- + + private uptimeSeconds(): number { + return Math.floor((this.now() - this.startedAt) / 1000); + } + + private within(xs: T[], ms: number): T[] { + const cutoff = this.now() - ms; + return xs.filter((x) => x.ts >= cutoff); + } + + /** Mean of the last RECENT_SAMPLES non-null values of `pick`. */ + private recentMean(pick: (r: RequestRecord) => number | null): number | null { + const vals: number[] = []; + for (let i = this.requests.length - 1; i >= 0 && vals.length < RECENT_SAMPLES; i--) { + const v = pick(this.requests[i]!); + if (v !== null) vals.push(v); + } + const m = mean(vals); + return m === null ? null : round1(m); + } + + /** Recent gen tok/s per model, over the last RECENT_SAMPLES samples each. */ + private tokSByModel(): Record { + const buckets = new Map(); + for (let i = this.requests.length - 1; i >= 0; i--) { + const r = this.requests[i]!; + if (r.genTokS === null) continue; + const arr = buckets.get(r.model) ?? []; + if (arr.length < RECENT_SAMPLES) { + arr.push(r.genTokS); + buckets.set(r.model, arr); + } + } + const out: Record = {}; + for (const [model, arr] of buckets) { + const m = mean(arr); + if (m !== null) out[model] = round1(m); + } + return out; + } + + private backendHealth(): Record { + const out: Record = {}; + for (const [model, h] of this.health) out[model] = h; + return out; + } + + private lastError(): { ts: string; model: ModelName | null; message: string } | null { + const e = this.errors[this.errors.length - 1]; + if (!e) return null; + return { ts: new Date(e.ts).toISOString(), model: e.model, message: e.message }; + } + + /** The telemetry-owned slice of `GET /status`. Queue fields are merged in by + * the dispatcher, which owns the queue and current model. */ + snapshot() { + return { + currentModelLoadedAt: + this.currentModelLoadedAt === null + ? null + : new Date(this.currentModelLoadedAt).toISOString(), + inFlight: this.inFlight, + lastSwapDurationMs: this.lastSwapDurationMs, + swapsLast15m: this.within(this.swaps, MS_15M).length, + swapsLastHour: this.within(this.swaps, MS_1H).length, + recentGenTokS: this.recentMean((r) => r.genTokS), + recentPromptTokS: this.recentMean((r) => r.promptTokS), + tokSByModel: this.tokSByModel(), + uptimeSeconds: this.uptimeSeconds(), + requestsServedTotal: this.requestsServedTotal, + requestsByModel: Object.fromEntries(this.requestsByModel), + tokensGeneratedTotal: this.tokensGeneratedTotal, + tokensByModel: Object.fromEntries(this.tokensByModel), + requestsPerMinute: this.within(this.requests, MS_1M).length, + errorsLast15m: this.within(this.errors, MS_15M).length, + lastError: this.lastError(), + backendHealth: this.backendHealth(), + }; + } + + /** Recent request records, newest first, for `GET /usage`. */ + usageList(limit: number): Array<{ + ts: string; + model: ModelName; + status: number; + total_ms: number; + ttfb_ms: number | null; + prompt_tokens: number | null; + gen_tokens: number | null; + gen_tok_s: number | null; + prompt_tok_s: number | null; + }> { + const n = Math.max(0, Math.min(limit, this.requests.length)); + const out = []; + for (let i = this.requests.length - 1; i >= this.requests.length - n; i--) { + const r = this.requests[i]!; + out.push({ + ts: new Date(r.ts).toISOString(), + model: r.model, + status: r.status, + total_ms: r.totalMs, + ttfb_ms: r.ttfbMs, + prompt_tokens: r.promptTokens, + gen_tokens: r.genTokens, + gen_tok_s: r.genTokS === null ? null : round1(r.genTokS), + prompt_tok_s: r.promptTokS === null ? null : round1(r.promptTokS), + }); + } + return out; + } + + /** + * Prometheus text exposition (v0.0.4). Queue gauges are passed in because + * the queue is owned elsewhere; everything else is from this aggregator. + */ + prometheus(queue: { + queueByModel: Record; + currentModel: ModelName | null; + }): string { + const L: string[] = []; + const esc = (s: string): string => s.replace(/\\/g, '\\\\').replace(/"/g, '\\"'); + + const help = (name: string, h: string, type: string): void => { + L.push(`# HELP ${name} ${h}`); + L.push(`# TYPE ${name} ${type}`); + }; + + help('jano_uptime_seconds', 'Seconds since jano started.', 'gauge'); + L.push(`jano_uptime_seconds ${this.uptimeSeconds()}`); + + help('jano_in_flight', 'Requests currently executing on a backend.', 'gauge'); + L.push(`jano_in_flight ${this.inFlight}`); + + help('jano_queue_depth', 'Requests waiting in the queue, per model.', 'gauge'); + for (const [model, n] of Object.entries(queue.queueByModel)) { + L.push(`jano_queue_depth{model="${esc(model)}"} ${n}`); + } + + help('jano_requests_served_total', 'Successfully served requests since start.', 'counter'); + for (const [model, n] of this.requestsByModel) { + L.push(`jano_requests_served_total{model="${esc(model)}"} ${n}`); + } + + help('jano_tokens_generated_total', 'Output tokens generated since start.', 'counter'); + for (const [model, n] of this.tokensByModel) { + L.push(`jano_tokens_generated_total{model="${esc(model)}"} ${n}`); + } + + help('jano_prompt_tokens_total', 'Prompt tokens processed since start.', 'counter'); + L.push(`jano_prompt_tokens_total ${this.promptTokensTotal}`); + + help('jano_recent_gen_tokens_per_second', 'Rolling generation tok/s, per model.', 'gauge'); + for (const [model, v] of Object.entries(this.tokSByModel())) { + L.push(`jano_recent_gen_tokens_per_second{model="${esc(model)}"} ${v}`); + } + + help('jano_errors_total', 'Errors recorded since start.', 'counter'); + L.push(`jano_errors_total ${this.errorsTotal}`); + + help('jano_swaps_total', 'Model swaps invoked since start, per transition.', 'counter'); + if (this.swapByTransition.size === 0) { + L.push(`jano_swaps_total ${this.swapsTotal}`); + } else { + for (const [key, agg] of this.swapByTransition) { + const [from, to] = key.split('→'); + L.push(`jano_swaps_total{from="${esc(from ?? '')}",to="${esc(to ?? '')}"} ${agg.count}`); + } + } + + // Per-transition cost table (swapCostByTransition): sum/count per + // modelA→modelB so "which transitions hurt" is derivable as sum/count. + help( + 'jano_swap_transition_duration_milliseconds', + 'Cumulative swap time per model transition.', + 'summary' + ); + for (const [key, agg] of this.swapByTransition) { + const [from, to] = key.split('→'); + const labels = `{from="${esc(from ?? '')}",to="${esc(to ?? '')}"}`; + L.push(`jano_swap_transition_duration_milliseconds_sum${labels} ${agg.totalMs}`); + L.push(`jano_swap_transition_duration_milliseconds_count${labels} ${agg.count}`); + } + + help('jano_swap_duration_milliseconds', 'Distribution of model-swap durations.', 'histogram'); + let cumulative = 0; + for (const le of SWAP_BUCKETS_MS) { + cumulative = this.swaps.filter((s) => s.durationMs <= le).length; + L.push(`jano_swap_duration_milliseconds_bucket{le="${le}"} ${cumulative}`); + } + L.push(`jano_swap_duration_milliseconds_bucket{le="+Inf"} ${this.swaps.length}`); + const swapSum = this.swaps.reduce((a, s) => a + s.durationMs, 0); + L.push(`jano_swap_duration_milliseconds_sum ${swapSum}`); + L.push(`jano_swap_duration_milliseconds_count ${this.swaps.length}`); + + help('jano_backend_up', 'Backend health: 1 up, 0 down/unknown.', 'gauge'); + for (const [model, h] of this.health) { + L.push(`jano_backend_up{model="${esc(model)}"} ${h === 'up' ? 1 : 0}`); + } + + return L.join('\n') + '\n'; + } +} diff --git a/tests/bodyTap.test.ts b/tests/bodyTap.test.ts new file mode 100644 index 0000000..3e3fd89 --- /dev/null +++ b/tests/bodyTap.test.ts @@ -0,0 +1,36 @@ +import { describe, expect, it } from 'vitest'; +import { Buffer } from 'node:buffer'; +import { BodyTap } from '../src/bodyTap.ts'; + +describe('BodyTap', () => { + it('non-streaming: accumulates the whole small body', () => { + const tap = new BodyTap(false); + tap.push(Buffer.from('{"usage":')); + tap.push(Buffer.from('{"completion_tokens":3}}')); + expect(tap.text()).toBe('{"usage":{"completion_tokens":3}}'); + }); + + it('non-streaming: returns null once the body overflows the cap', () => { + const tap = new BodyTap(false, 8); + tap.push(Buffer.from('1234')); + tap.push(Buffer.from('56789')); // pushes past cap of 8 + expect(tap.text()).toBeNull(); + }); + + it('streaming: keeps only the tail within cap, preserving the final frame', () => { + const tap = new BodyTap(true, 16); + tap.push(Buffer.from('AAAAAAAAAAAAAAAA')); // 16 bytes, evicted by later pushes + tap.push(Buffer.from('data: {"x":1}\n')); + const text = tap.text(); + expect(text).not.toBeNull(); + expect(text).toContain('data: {"x":1}'); + expect(text).not.toContain('AAAA'); + }); + + it('streaming: a single oversized frame is still retained', () => { + const tap = new BodyTap(true, 4); + const big = 'data: {"usage":{"completion_tokens":42}}'; + tap.push(Buffer.from(big)); + expect(tap.text()).toBe(big); + }); +}); diff --git a/tests/responseStats.test.ts b/tests/responseStats.test.ts new file mode 100644 index 0000000..8d3cd75 --- /dev/null +++ b/tests/responseStats.test.ts @@ -0,0 +1,145 @@ +import { describe, expect, it } from 'vitest'; +import { parseResponseStats } from '../src/responseStats.ts'; + +describe('parseResponseStats (non-streaming JSON)', () => { + it('reads OpenAI usage counts', () => { + const body = JSON.stringify({ + choices: [{ message: { content: 'hi' } }], + usage: { prompt_tokens: 12, completion_tokens: 34, total_tokens: 46 }, + }); + expect(parseResponseStats(body, false)).toEqual({ + promptTokens: 12, + genTokens: 34, + genTokS: null, + promptTokS: null, + }); + }); + + it('reads llama.cpp timings (tok/s come from the backend)', () => { + const body = JSON.stringify({ + choices: [{ message: { content: 'hi' } }], + timings: { + prompt_n: 20, + prompt_per_second: 250.5, + predicted_n: 100, + predicted_per_second: 47.2, + }, + }); + expect(parseResponseStats(body, false)).toEqual({ + promptTokens: 20, + genTokens: 100, + genTokS: 47.2, + promptTokS: 250.5, + }); + }); + + it('prefers usage counts but still takes tok/s from timings', () => { + const body = JSON.stringify({ + usage: { prompt_tokens: 5, completion_tokens: 6 }, + timings: { + prompt_n: 999, + predicted_n: 999, + predicted_per_second: 40, + prompt_per_second: 200, + }, + }); + expect(parseResponseStats(body, false)).toEqual({ + promptTokens: 5, + genTokens: 6, + genTokS: 40, + promptTokS: 200, + }); + }); + + it('normalizes Ollama-style nanosecond timings to tok/s', () => { + const body = JSON.stringify({ + message: { content: 'hi' }, + prompt_eval_count: 10, + prompt_eval_duration: 50_000_000, // 50ms → 200 tok/s + eval_count: 20, + eval_duration: 500_000_000, // 500ms → 40 tok/s + }); + expect(parseResponseStats(body, false)).toEqual({ + promptTokens: 10, + genTokens: 20, + genTokS: 40, + promptTokS: 200, + }); + }); + + it('ignores Ollama-style zero durations rather than dividing by zero', () => { + const body = JSON.stringify({ eval_count: 5, eval_duration: 0 }); + expect(parseResponseStats(body, false)).toEqual({ + promptTokens: null, + genTokens: 5, + genTokS: null, + promptTokS: null, + }); + }); + + it('returns null when neither usage nor timings present', () => { + expect(parseResponseStats(JSON.stringify({ choices: [] }), false)).toBeNull(); + }); + + it('returns null on invalid JSON', () => { + expect(parseResponseStats('not json', false)).toBeNull(); + }); + + it('ignores non-finite numbers', () => { + const body = JSON.stringify({ usage: { prompt_tokens: 'x', completion_tokens: null } }); + expect(parseResponseStats(body, false)).toBeNull(); + }); +}); + +describe('parseResponseStats (SSE streaming)', () => { + it('finds usage in the final chunk (include_usage)', () => { + const sse = [ + 'data: {"choices":[{"delta":{"content":"he"}}]}', + '', + 'data: {"choices":[{"delta":{"content":"llo"}}]}', + '', + 'data: {"choices":[],"usage":{"prompt_tokens":7,"completion_tokens":2}}', + '', + 'data: [DONE]', + '', + ].join('\n'); + expect(parseResponseStats(sse, true)).toEqual({ + promptTokens: 7, + genTokens: 2, + genTokS: null, + promptTokS: null, + }); + }); + + it('finds llama.cpp timings in the trailing frame', () => { + const sse = [ + 'data: {"choices":[{"delta":{"content":"x"}}]}', + '', + 'data: {"choices":[],"timings":{"predicted_n":3,"predicted_per_second":51.0,"prompt_n":4,"prompt_per_second":120}}', + '', + 'data: [DONE]', + ].join('\n'); + expect(parseResponseStats(sse, true)).toEqual({ + promptTokens: 4, + genTokens: 3, + genTokS: 51.0, + promptTokS: 120, + }); + }); + + it('returns null for a stream that never reports usage', () => { + const sse = ['data: {"choices":[{"delta":{"content":"x"}}]}', '', 'data: [DONE]'].join('\n'); + expect(parseResponseStats(sse, true)).toBeNull(); + }); + + it('tolerates unparseable frames and keeps scanning', () => { + const sse = [ + 'data: {bogus', + '', + 'data: {"usage":{"prompt_tokens":1,"completion_tokens":9}}', + '', + 'data: [DONE]', + ].join('\n'); + expect(parseResponseStats(sse, true)?.genTokens).toBe(9); + }); +}); diff --git a/tests/telemetry-e2e.test.ts b/tests/telemetry-e2e.test.ts new file mode 100644 index 0000000..1260762 --- /dev/null +++ b/tests/telemetry-e2e.test.ts @@ -0,0 +1,258 @@ +import { afterAll, beforeAll, describe, expect, it } from 'vitest'; +import { spawn, type ChildProcessByStdio } from 'node:child_process'; +import type { Readable } from 'node:stream'; +import { mkdtempSync, rmSync, writeFileSync } from 'node:fs'; +import http from 'node:http'; +import { tmpdir } from 'node:os'; +import { join, resolve } from 'node:path'; +import { setTimeout as sleep } from 'node:timers/promises'; +import { request as undiciRequest } from 'undici'; + +// End-to-end coverage for the telemetry surface (the Ollama-gap "wedge" +// metrics): /status extras, /usage records, and /metrics Prometheus output, +// exercised against the real jano binary with two mock backends so a swap is +// actually recorded. + +const HOST = '127.0.0.1'; +let JANO_PORT: number; +let ALPHA_PORT: number; +let BETA_PORT: number; +let JANO_BASE: string; +let tmpDir: string; + +// A mock OpenAI/llama.cpp backend: /health 200; non-streaming chat returns a +// body carrying both `usage` and llama.cpp `timings`; streaming chat emits a +// couple of SSE deltas then a final frame with usage + timings. +function startBackend(capture: (port: number) => void): Promise { + const server = http.createServer((req, res) => { + if (req.url === '/health') { + res.writeHead(200, { 'Content-Type': 'text/plain' }); + res.end('ok'); + return; + } + if (req.method === 'POST' && req.url === '/v1/chat/completions') { + const chunks: Buffer[] = []; + req.on('data', (c: Buffer) => chunks.push(c)); + req.on('end', () => { + const body = Buffer.concat(chunks).toString('utf8'); + const stream = body.includes('"stream":true') || body.includes('"stream": true'); + if (stream) { + res.writeHead(200, { 'Content-Type': 'text/event-stream' }); + res.write('data: {"choices":[{"delta":{"content":"hel"}}]}\n\n'); + res.write('data: {"choices":[{"delta":{"content":"lo"}}]}\n\n'); + res.write( + 'data: {"choices":[],"usage":{"prompt_tokens":7,"completion_tokens":13},"timings":{"predicted_n":13,"predicted_per_second":55.5,"prompt_n":7,"prompt_per_second":300}}\n\n' + ); + res.end('data: [DONE]\n\n'); + return; + } + res.writeHead(200, { 'Content-Type': 'application/json' }); + res.end( + JSON.stringify({ + choices: [{ message: { content: 'hi' } }], + usage: { prompt_tokens: 11, completion_tokens: 22, total_tokens: 33 }, + timings: { + prompt_n: 11, + prompt_per_second: 300, + predicted_n: 22, + predicted_per_second: 42, + }, + }) + ); + }); + return; + } + res.writeHead(404); + res.end(); + }); + return new Promise((r, reject) => { + server.once('error', reject); + server.listen(0, HOST, () => { + const addr = server.address(); + if (addr === null || typeof addr === 'string') { + reject(new Error('backend did not bind')); + return; + } + capture(addr.port); + r(server); + }); + }); +} + +async function pickFreePort(): Promise { + const probe = http.createServer(); + return new Promise((r, reject) => { + probe.once('error', reject); + probe.listen(0, HOST, () => { + const addr = probe.address(); + if (addr === null || typeof addr === 'string') { + reject(new Error('probe did not bind')); + return; + } + const port = addr.port; + probe.close(() => r(port)); + }); + }); +} + +function startJano(): ChildProcessByStdio { + const cwd = resolve(import.meta.dirname, '..'); + const modelsFile = join(tmpDir, 'models.json'); + writeFileSync( + modelsFile, + JSON.stringify({ + models: [ + { name: 'alpha', url: `http://${HOST}:${ALPHA_PORT}` }, + { name: 'beta', url: `http://${HOST}:${BETA_PORT}` }, + ], + }) + ); + return spawn(process.execPath, ['--experimental-strip-types', 'src/index.ts'], { + cwd, + env: { + ...process.env, + JANO_HOST: HOST, + JANO_PORT: String(JANO_PORT), + SWAP_COMMAND: resolve(cwd, 'tests/fixtures/swap-noop.sh'), + MODELS_FILE: modelsFile, + HEALTH_POLL_INTERVAL_MS: '50', + }, + stdio: ['ignore', 'pipe', 'pipe'], + }); +} + +async function waitForJanoReady(timeoutMs: number): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + try { + const res = await fetch(`${JANO_BASE}/v1/models`); + if (res.ok) return; + } catch { + // Not up yet. + } + await sleep(50); + } + throw new Error(`jano did not start within ${timeoutMs}ms`); +} + +async function postChat(model: string, stream = false): Promise { + const { statusCode, body } = await undiciRequest(`${JANO_BASE}/v1/chat/completions`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ model, stream, messages: [{ role: 'user', content: 'ping' }] }), + }); + // Drain so the request fully completes before we read telemetry. + await body.text(); + return statusCode; +} + +async function getJson(path: string): Promise<{ status: number; json: Record }> { + const { statusCode, body } = await undiciRequest(`${JANO_BASE}${path}`); + const text = await body.text(); + return { status: statusCode, json: JSON.parse(text) as Record }; +} + +describe('telemetry surface (Ollama-gap wedge metrics)', () => { + let alpha: http.Server; + let beta: http.Server; + let jano: ChildProcessByStdio; + const janoStdout: string[] = []; + + beforeAll(async () => { + tmpDir = mkdtempSync(join(tmpdir(), 'jano-telemetry-')); + alpha = await startBackend((p) => { + ALPHA_PORT = p; + }); + beta = await startBackend((p) => { + BETA_PORT = p; + }); + JANO_PORT = await pickFreePort(); + JANO_BASE = `http://${HOST}:${JANO_PORT}`; + jano = startJano(); + jano.stdout.on('data', (b: Buffer) => janoStdout.push(b.toString('utf8'))); + jano.stderr.on('data', (b: Buffer) => janoStdout.push(b.toString('utf8'))); + await waitForJanoReady(10_000); + + // Drive traffic: alpha (initial), then beta (forces one swap), then a + // streaming beta request (exercises the SSE body tap). + expect(await postChat('alpha')).toBe(200); + expect(await postChat('beta')).toBe(200); + expect(await postChat('beta', true)).toBe(200); + }, 25_000); + + afterAll(async () => { + if (jano && jano.exitCode === null) { + jano.kill('SIGTERM'); + await new Promise((r) => jano.once('exit', () => r())); + } + if (alpha) await new Promise((r) => alpha.close(() => r())); + if (beta) await new Promise((r) => beta.close(() => r())); + if (tmpDir) rmSync(tmpDir, { recursive: true, force: true }); + }); + + it('GET /status carries queue, swap, throughput, and counter telemetry', async () => { + const { status, json } = await getJson('/status'); + expect(status).toBe(200); + expect(json.ok).toBe(true); + expect(json.currentModel).toBe('beta'); + + // Queue / swap economics. + expect(json).toHaveProperty('queueByModel'); + expect(json.inFlight).toBe(0); + expect(json.oldestWaitingMs).toBe(0); + expect(typeof json.lastSwapDurationMs).toBe('number'); + expect(json.swapsLastHour).toBeGreaterThanOrEqual(1); + expect(typeof json.currentModelLoadedAt).toBe('string'); + + // Throughput + cumulative counters. + expect(json.requestsServedTotal).toBe(3); + expect(json.tokensGeneratedTotal).toBe(22 + 22 + 13); + expect(json.recentGenTokS).toBeGreaterThan(0); + expect(json.uptimeSeconds).toBeGreaterThanOrEqual(0); + const tokS = json.tokSByModel as Record; + expect(typeof tokS.alpha).toBe('number'); + expect(typeof tokS.beta).toBe('number'); + + // Health derived from successful swaps + serves. + expect(json.backendHealth).toMatchObject({ alpha: 'up', beta: 'up' }); + expect(json.errorsLast15m).toBe(0); + expect(json.lastError).toBeNull(); + }); + + it('GET /usage returns recent per-request records, newest first', async () => { + const { status, json } = await getJson('/usage?limit=10'); + expect(status).toBe(200); + expect(json.count).toBe(3); + const records = json.records as Array>; + // Newest first: the streaming beta request. + expect(records[0]!.model).toBe('beta'); + expect(records[0]!.gen_tokens).toBe(13); + expect(records[0]!.gen_tok_s).toBe(55.5); + // Every record has the spec'd usage-record shape. + for (const r of records) { + expect(r).toHaveProperty('ts'); + expect(r).toHaveProperty('status', 200); + expect(r).toHaveProperty('total_ms'); + expect(r).toHaveProperty('prompt_tokens'); + } + }); + + it('GET /usage honours the limit', async () => { + const { json } = await getJson('/usage?limit=1'); + expect(json.count).toBe(1); + }); + + it('GET /metrics emits Prometheus exposition', async () => { + const { statusCode, headers, body } = await undiciRequest(`${JANO_BASE}/metrics`); + const text = await body.text(); + expect(statusCode).toBe(200); + expect(String(headers['content-type'])).toContain('text/plain'); + expect(text).toContain('# TYPE jano_uptime_seconds gauge'); + expect(text).toContain('jano_requests_served_total{model="alpha"}'); + expect(text).toContain('jano_requests_served_total{model="beta"}'); + expect(text).toMatch(/jano_tokens_generated_total\{model="beta"\} \d+/); + expect(text).toContain('jano_swap_duration_milliseconds_count 1'); + expect(text).toMatch(/jano_swaps_total\{from="alpha",to="beta"\} 1/); + expect(text).toContain('jano_backend_up{model="alpha"} 1'); + }); +}); diff --git a/tests/telemetry.test.ts b/tests/telemetry.test.ts new file mode 100644 index 0000000..1a224be --- /dev/null +++ b/tests/telemetry.test.ts @@ -0,0 +1,201 @@ +import { beforeEach, describe, expect, it } from 'vitest'; +import { Telemetry, type RequestRecord } from '../src/telemetry.ts'; + +let now = 0; +const clock = () => now; + +function req(over: Partial> = {}): Omit { + return { + model: 'chat', + status: 200, + totalMs: 100, + ttfbMs: 10, + promptTokens: 5, + genTokens: 50, + genTokS: 40, + promptTokS: 200, + ...over, + }; +} + +describe('Telemetry', () => { + beforeEach(() => { + now = 1_000_000; + }); + + describe('counters', () => { + it('counts only 2xx toward served-total and tokens', () => { + const t = new Telemetry({ now: clock }); + t.recordRequest(req({ genTokens: 50 })); + t.recordRequest(req({ genTokens: 30 })); + t.recordRequest(req({ status: 503, genTokens: null })); + const s = t.snapshot(); + expect(s.requestsServedTotal).toBe(2); + expect(s.tokensGeneratedTotal).toBe(80); + expect(s.requestsByModel).toEqual({ chat: 2 }); + expect(s.tokensByModel).toEqual({ chat: 80 }); + }); + + it('non-2xx still lands in the usage history', () => { + const t = new Telemetry({ now: clock }); + t.recordRequest(req({ status: 503, genTokens: null })); + const list = t.usageList(10); + expect(list).toHaveLength(1); + expect(list[0]!.status).toBe(503); + }); + }); + + describe('rolling tok/s', () => { + it('averages recent gen + prompt tok/s', () => { + const t = new Telemetry({ now: clock }); + t.recordRequest(req({ genTokS: 40, promptTokS: 200 })); + t.recordRequest(req({ genTokS: 60, promptTokS: 100 })); + const s = t.snapshot(); + expect(s.recentGenTokS).toBe(50); + expect(s.recentPromptTokS).toBe(150); + }); + + it('skips records with null tok/s', () => { + const t = new Telemetry({ now: clock }); + t.recordRequest(req({ genTokS: null })); + expect(t.snapshot().recentGenTokS).toBeNull(); + }); + + it('reports tok/s per model', () => { + const t = new Telemetry({ now: clock }); + t.recordRequest(req({ model: 'chat', genTokS: 50 })); + t.recordRequest(req({ model: 'code', genTokS: 30 })); + expect(t.snapshot().tokSByModel).toEqual({ chat: 50, code: 30 }); + }); + }); + + describe('swaps', () => { + it('records duration, loadedAt, and per-window counts', () => { + const t = new Telemetry({ now: clock }); + t.recordSwap(null, 'chat', 21_840); + const s = t.snapshot(); + expect(s.lastSwapDurationMs).toBe(21_840); + expect(s.currentModelLoadedAt).toBe(new Date(now).toISOString()); + expect(s.swapsLast15m).toBe(1); + expect(s.swapsLastHour).toBe(1); + }); + + it('ages swaps out of the 15m / 1h windows', () => { + const t = new Telemetry({ now: clock }); + t.recordSwap('chat', 'code', 1000); // at t0 + now += 20 * 60_000; // +20 min + t.recordSwap('code', 'chat', 1000); // recent + const s = t.snapshot(); + expect(s.swapsLast15m).toBe(1); // only the +20m one is within 15m of now + expect(s.swapsLastHour).toBe(2); // both within the hour + }); + }); + + describe('errors + health', () => { + it('tracks recent errors, last error, and marks the backend down', () => { + const t = new Telemetry({ now: clock }); + t.seedModels(['chat', 'code']); + t.recordError('code', 'swap failed: boom'); + const s = t.snapshot(); + expect(s.errorsLast15m).toBe(1); + expect(s.lastError?.model).toBe('code'); + expect(s.lastError?.message).toContain('boom'); + expect(s.backendHealth).toEqual({ chat: 'unknown', code: 'down' }); + }); + + it('a later successful request flips health back up', () => { + const t = new Telemetry({ now: clock }); + t.recordError('code', 'boom'); + t.recordRequest(req({ model: 'code', status: 200 })); + expect(t.snapshot().backendHealth.code).toBe('up'); + }); + }); + + describe('in-flight', () => { + it('increments and decrements, never below zero', () => { + const t = new Telemetry({ now: clock }); + t.incInFlight(); + t.incInFlight(); + expect(t.snapshot().inFlight).toBe(2); + t.decInFlight(); + t.decInFlight(); + t.decInFlight(); + expect(t.snapshot().inFlight).toBe(0); + }); + }); + + describe('usage ring', () => { + it('caps at usageMax, dropping oldest', () => { + const t = new Telemetry({ now: clock, usageMax: 3 }); + for (let i = 0; i < 5; i++) t.recordRequest(req({ totalMs: i })); + const list = t.usageList(100); + expect(list).toHaveLength(3); + // newest first: totalMs 4, 3, 2 + expect(list.map((r) => r.total_ms)).toEqual([4, 3, 2]); + }); + + it('emits snake_case fields with ISO timestamps', () => { + const t = new Telemetry({ now: clock }); + t.recordRequest(req()); + const [rec] = t.usageList(1); + expect(rec).toMatchObject({ + model: 'chat', + status: 200, + total_ms: 100, + ttfb_ms: 10, + prompt_tokens: 5, + gen_tokens: 50, + gen_tok_s: 40, + prompt_tok_s: 200, + }); + expect(rec!.ts).toBe(new Date(now).toISOString()); + }); + }); + + describe('prometheus exposition', () => { + it('emits counters, gauges, and a swap histogram', () => { + const t = new Telemetry({ now: clock }); + t.seedModels(['chat', 'code']); + t.recordSwap(null, 'chat', 12_000); + t.recordRequest(req({ model: 'chat', genTokens: 50 })); + t.recordError('code', 'boom'); + const text = t.prometheus({ queueByModel: { chat: 1, code: 0 }, currentModel: 'chat' }); + + expect(text).toContain('# TYPE jano_uptime_seconds gauge'); + expect(text).toContain('jano_requests_served_total{model="chat"} 1'); + expect(text).toContain('jano_tokens_generated_total{model="chat"} 50'); + expect(text).toContain('jano_queue_depth{model="chat"} 1'); + expect(text).toContain('jano_errors_total 1'); + expect(text).toContain('jano_swaps_total{from="(none)",to="chat"} 1'); + // Per-transition cost table (swapCostByTransition). + expect(text).toContain( + 'jano_swap_transition_duration_milliseconds_sum{from="(none)",to="chat"} 12000' + ); + expect(text).toContain( + 'jano_swap_transition_duration_milliseconds_count{from="(none)",to="chat"} 1' + ); + expect(text).toContain('jano_backend_up{model="chat"} 1'); + expect(text).toContain('jano_backend_up{model="code"} 0'); + // swap of 12s lands in the 15000ms bucket but not the 5000ms one + expect(text).toContain('jano_swap_duration_milliseconds_bucket{le="5000"} 0'); + expect(text).toContain('jano_swap_duration_milliseconds_bucket{le="15000"} 1'); + expect(text).toContain('jano_swap_duration_milliseconds_count 1'); + expect(text).toMatch(/jano_swap_duration_milliseconds_sum 12000/); + }); + + it('escapes quotes/backslashes in label values', () => { + const t = new Telemetry({ now: clock }); + t.recordRequest(req({ model: 'we"ird' })); + const text = t.prometheus({ queueByModel: {}, currentModel: null }); + expect(text).toContain('jano_requests_served_total{model="we\\"ird"} 1'); + }); + }); + + describe('uptime', () => { + it('reports whole seconds since construction', () => { + const t = new Telemetry({ now: clock }); + now += 84_211_000; + expect(t.snapshot().uptimeSeconds).toBe(84_211); + }); + }); +}); From a96eb312e086b06f0104cb161dbffcfa0cae7fb9 Mon Sep 17 00:00:00 2001 From: Leonardo Cardoso Date: Fri, 19 Jun 2026 23:35:04 +0200 Subject: [PATCH 2/2] chore: v0.2.0 --- package.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/package.json b/package.json index 9c6d9ed..412e23c 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "jano", - "version": "0.1.0", + "version": "0.2.0", "description": "OpenAI-compatible router that batches LLM requests by model so a flurry of mixed calls costs one swap, not many. Backend-agnostic; works with any number of models.", "private": true, "type": "module",