:)
This commit is contained in:
+37
@@ -4,6 +4,7 @@ import {
|
||||
createSession,
|
||||
deleteSession,
|
||||
getSession,
|
||||
getSetting,
|
||||
getUserById,
|
||||
getUserByUsername,
|
||||
toPublicUser,
|
||||
@@ -109,3 +110,39 @@ export function recordAttempt(ip: string): void {
|
||||
export function clearAttempts(ip: string): void {
|
||||
attempts.delete(ip);
|
||||
}
|
||||
|
||||
// --- dedicated rate limiter for password recovery (admin-configurable) ---
|
||||
// Separate bucket from login: recovery is anonymous-by-design (always-ok
|
||||
// answers) so it needs its own, usually stricter, envelope.
|
||||
|
||||
const recoveryAttempts = new Map<string, number[]>();
|
||||
|
||||
function clampSettingInt(raw: string, min: number, max: number, fallback: number): number {
|
||||
const v = Number.parseInt(raw, 10);
|
||||
if (!Number.isInteger(v)) return fallback;
|
||||
return Math.min(max, Math.max(min, v));
|
||||
}
|
||||
|
||||
export async function recoveryLimitMax(): Promise<number> {
|
||||
return clampSettingInt(await getSetting("recovery_limit_max", "5"), 1, 100, 5);
|
||||
}
|
||||
|
||||
export async function recoveryLimitWindowMs(): Promise<number> {
|
||||
return clampSettingInt(await getSetting("recovery_limit_window_min", "60"), 1, 1440, 60) * 60_000;
|
||||
}
|
||||
|
||||
export async function isRecoveryLimited(ip: string): Promise<{ limited: boolean; retryAfterSec: number }> {
|
||||
const windowMs = await recoveryLimitWindowMs();
|
||||
const max = await recoveryLimitMax();
|
||||
const now = Date.now();
|
||||
const list = (recoveryAttempts.get(ip) ?? []).filter((t) => t > now - windowMs);
|
||||
recoveryAttempts.set(ip, list);
|
||||
if (list.length < max) return { limited: false, retryAfterSec: 0 };
|
||||
return { limited: true, retryAfterSec: Math.max(1, Math.ceil((list[0]! + windowMs - now) / 1000)) };
|
||||
}
|
||||
|
||||
export function recordRecoveryAttempt(ip: string): void {
|
||||
const list = recoveryAttempts.get(ip) ?? [];
|
||||
list.push(Date.now());
|
||||
recoveryAttempts.set(ip, list);
|
||||
}
|
||||
|
||||
+99
-23
@@ -1,12 +1,28 @@
|
||||
import { config } from "../config.ts";
|
||||
import { driverLog, fmtBytes, fmtMs } from "./log.ts";
|
||||
import {
|
||||
effectiveOllamaNodes,
|
||||
getOllamaTimeoutMs,
|
||||
pickOllamaRoute,
|
||||
recordOllamaStat,
|
||||
type EffectiveNode,
|
||||
} from "../ollamaNodes.ts";
|
||||
import type { ChatMessage, ChatOptions, DriverModel, LLMDriver } from "./types.ts";
|
||||
|
||||
interface OllamaTag {
|
||||
name: string;
|
||||
}
|
||||
|
||||
/** Local LLM via Ollama. Priority #1. */
|
||||
interface OllamaDone {
|
||||
done?: boolean;
|
||||
error?: string;
|
||||
message?: { content?: string };
|
||||
prompt_eval_count?: number;
|
||||
eval_count?: number;
|
||||
eval_duration?: number;
|
||||
}
|
||||
|
||||
/** Local LLM via Ollama (multi-node pool). Priority #1. */
|
||||
export class OllamaDriver implements LLMDriver {
|
||||
readonly name = "ollama";
|
||||
readonly enabled: boolean;
|
||||
@@ -16,22 +32,31 @@ export class OllamaDriver implements LLMDriver {
|
||||
this.enabled = enabled;
|
||||
}
|
||||
|
||||
/** Model list merged across every enabled node (deduped by name). */
|
||||
async listModels(): Promise<DriverModel[]> {
|
||||
const t0 = Date.now();
|
||||
const res = await fetch(`${this.host}/api/tags`, { signal: AbortSignal.timeout(8000) });
|
||||
if (!res.ok) {
|
||||
driverLog(this.name, `listModels ERROR: /api/tags -> ${res.status} (${fmtMs(Date.now() - t0)})`);
|
||||
throw new Error(`ollama /api/tags failed: ${res.status}`);
|
||||
}
|
||||
const data = (await res.json()) as { models?: OllamaTag[] };
|
||||
const models = data.models ?? [];
|
||||
driverLog(this.name, `listModels -> ${models.length} model(s) [${models.map((m) => m.name).join(", ")}] (${fmtMs(Date.now() - t0)})`);
|
||||
return models.map((m) => ({
|
||||
id: `ollama:${m.name}`,
|
||||
name: m.name,
|
||||
driver: "ollama",
|
||||
label: m.name,
|
||||
const nodes = await effectiveOllamaNodes();
|
||||
if (nodes.length === 0) throw new Error("no ollama node available (all disabled or weight 0)");
|
||||
const seen = new Map<string, DriverModel>();
|
||||
const errors: string[] = [];
|
||||
await Promise.all(nodes.map(async (n) => {
|
||||
try {
|
||||
const res = await fetch(`${n.host}/api/tags`, { signal: AbortSignal.timeout(8000) });
|
||||
if (!res.ok) throw new Error(`status ${res.status}`);
|
||||
const data = (await res.json()) as { models?: OllamaTag[] };
|
||||
for (const m of data.models ?? []) {
|
||||
if (typeof m.name === "string" && !seen.has(m.name)) {
|
||||
seen.set(m.name, { id: `ollama:${m.name}`, name: m.name, driver: "ollama", label: m.name });
|
||||
}
|
||||
}
|
||||
} catch (e) {
|
||||
errors.push(`${n.name}: ${(e as Error).message}`);
|
||||
}
|
||||
}));
|
||||
const models = [...seen.values()];
|
||||
driverLog(this.name, `listModels -> ${models.length} model(s) across ${nodes.length} node(s) (${fmtMs(Date.now() - t0)})${errors.length > 0 ? ` [skipped: ${errors.join("; ")}]` : ""}`);
|
||||
if (models.length === 0 && errors.length > 0) throw new Error(`ollama listModels failed: ${errors.join("; ")}`);
|
||||
return models;
|
||||
}
|
||||
|
||||
async chat(messages: ChatMessage[], opts: ChatOptions): Promise<string> {
|
||||
@@ -41,17 +66,56 @@ export class OllamaDriver implements LLMDriver {
|
||||
}
|
||||
|
||||
async *chatStream(messages: ChatMessage[], opts: ChatOptions): AsyncGenerator<string, void, void> {
|
||||
const timeoutMs = await getOllamaTimeoutMs();
|
||||
const { order } = await pickOllamaRoute();
|
||||
let lastError: Error | null = null;
|
||||
for (const node of order) {
|
||||
// Failover only while nothing was produced yet: once tokens reached
|
||||
// the client, restarting on another node would duplicate the reply.
|
||||
let produced = false;
|
||||
const counting: ChatOptions = {
|
||||
...opts,
|
||||
onToken: (t) => {
|
||||
produced = true;
|
||||
opts.onToken?.(t);
|
||||
},
|
||||
};
|
||||
try {
|
||||
yield* this.streamFromNode(node, messages, counting, timeoutMs);
|
||||
return;
|
||||
} catch (e) {
|
||||
// Caller hangup: never fail over (nobody is listening anymore).
|
||||
if (opts.signal?.aborted) throw e;
|
||||
if (produced) throw e;
|
||||
lastError = e as Error;
|
||||
driverLog(this.name, `chat model=${opts.model} node ${node.name} failed, failing over: ${lastError.message}`);
|
||||
}
|
||||
}
|
||||
throw lastError ?? new Error("ollama chat failed on every node");
|
||||
}
|
||||
|
||||
/** One streaming attempt against a single node (throws on any failure). */
|
||||
private async *streamFromNode(
|
||||
node: EffectiveNode,
|
||||
messages: ChatMessage[],
|
||||
opts: ChatOptions,
|
||||
timeoutMs: number,
|
||||
): AsyncGenerator<string, void, void> {
|
||||
const t0 = Date.now();
|
||||
driverLog(this.name, `chat model=${opts.model} messages=${messages.length} stream=open`);
|
||||
const res = await fetch(`${this.host}/api/chat`, {
|
||||
driverLog(this.name, `chat model=${opts.model} node=${node.name} stream=open`);
|
||||
const signals = [AbortSignal.timeout(timeoutMs)];
|
||||
if (opts.signal) signals.push(opts.signal);
|
||||
const signal = typeof AbortSignal.any === "function" ? AbortSignal.any(signals) : signals[0]!;
|
||||
const res = await fetch(`${node.host}/api/chat`, {
|
||||
method: "POST",
|
||||
headers: { "Content-Type": "application/json" },
|
||||
body: JSON.stringify({ model: opts.model, messages, stream: true }),
|
||||
signal: opts.signal,
|
||||
signal,
|
||||
});
|
||||
if (!res.ok || !res.body) {
|
||||
const text = await res.text().catch(() => "");
|
||||
driverLog(this.name, `chat model=${opts.model} ERROR: upstream ${res.status} ${text.slice(0, 120)}`);
|
||||
driverLog(this.name, `chat model=${opts.model} node=${node.name} ERROR: upstream ${res.status} ${text.slice(0, 120)}`);
|
||||
await recordOllamaStat({ node_id: node.id, ok: false, latency_ms: Date.now() - t0, prompt_tokens: 0, eval_tokens: 0, model: opts.model });
|
||||
throw new Error(`ollama chat failed (${res.status}): ${text.slice(0, 300)}`);
|
||||
}
|
||||
const reader = res.body.getReader();
|
||||
@@ -59,6 +123,8 @@ export class OllamaDriver implements LLMDriver {
|
||||
let buf = "";
|
||||
let chunks = 0;
|
||||
let chars = 0;
|
||||
let promptTokens = 0;
|
||||
let evalTokens = 0;
|
||||
let status = "done";
|
||||
try {
|
||||
while (true) {
|
||||
@@ -71,7 +137,7 @@ export class OllamaDriver implements LLMDriver {
|
||||
const trimmed = line.trim();
|
||||
if (!trimmed) continue;
|
||||
try {
|
||||
const obj = JSON.parse(trimmed) as { message?: { content?: string }; error?: string; done?: boolean };
|
||||
const obj = JSON.parse(trimmed) as OllamaDone;
|
||||
if (obj.error) throw new Error(`ollama error: ${obj.error}`);
|
||||
const token = obj.message?.content ?? "";
|
||||
if (token) {
|
||||
@@ -80,8 +146,11 @@ export class OllamaDriver implements LLMDriver {
|
||||
opts.onToken?.(token);
|
||||
yield token;
|
||||
}
|
||||
if (typeof obj.prompt_eval_count === "number") promptTokens = obj.prompt_eval_count;
|
||||
if (typeof obj.eval_count === "number") evalTokens = obj.eval_count;
|
||||
if (obj.done) {
|
||||
driverLog(this.name, `chat model=${opts.model} done: ${chunks} chunks, ${fmtBytes(chars)} in ${fmtMs(Date.now() - t0)}`);
|
||||
driverLog(this.name, `chat model=${opts.model} node=${node.name} done: ${chunks} chunks, ${fmtBytes(chars)} in ${fmtMs(Date.now() - t0)}`);
|
||||
await recordOllamaStat({ node_id: node.id, ok: true, latency_ms: Date.now() - t0, prompt_tokens: promptTokens, eval_tokens: evalTokens, model: opts.model });
|
||||
return;
|
||||
}
|
||||
} catch (e) {
|
||||
@@ -90,9 +159,16 @@ export class OllamaDriver implements LLMDriver {
|
||||
}
|
||||
}
|
||||
}
|
||||
driverLog(this.name, `chat model=${opts.model} upstream closed: ${chunks} chunks, ${fmtBytes(chars)} in ${fmtMs(Date.now() - t0)}`);
|
||||
driverLog(this.name, `chat model=${opts.model} node=${node.name} upstream closed: ${chunks} chunks, ${fmtBytes(chars)} in ${fmtMs(Date.now() - t0)}`);
|
||||
// Truncated stream (no done:true): still a usable partial reply.
|
||||
await recordOllamaStat({ node_id: node.id, ok: true, latency_ms: Date.now() - t0, prompt_tokens: promptTokens, eval_tokens: evalTokens, model: opts.model });
|
||||
} catch (e) {
|
||||
status = (e as Error).name === "AbortError" ? "aborted-by-caller" : `ERROR: ${(e as Error).message}`;
|
||||
const name = (e as Error).name;
|
||||
status = name === "AbortError" ? "aborted-by-caller" : name === "TimeoutError" ? "timeout" : `ERROR: ${(e as Error).message}`;
|
||||
// Caller aborts are not node failures — don't pollute telemetry.
|
||||
if (name !== "AbortError" || !opts.signal?.aborted) {
|
||||
await recordOllamaStat({ node_id: node.id, ok: false, latency_ms: Date.now() - t0, prompt_tokens: promptTokens, eval_tokens: evalTokens, model: opts.model });
|
||||
}
|
||||
throw e;
|
||||
} finally {
|
||||
// Stop the upstream generation when the consumer goes away
|
||||
@@ -104,7 +180,7 @@ export class OllamaDriver implements LLMDriver {
|
||||
}
|
||||
reader.releaseLock();
|
||||
if (status !== "done") {
|
||||
driverLog(this.name, `chat model=${opts.model} ${status} after ${chunks} chunks, ${fmtBytes(chars)} in ${fmtMs(Date.now() - t0)}`);
|
||||
driverLog(this.name, `chat model=${opts.model} node=${node.name} ${status} after ${chunks} chunks, ${fmtBytes(chars)} in ${fmtMs(Date.now() - t0)}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+264
-2
@@ -5,14 +5,35 @@ import {
|
||||
extractBearer,
|
||||
hashSecret,
|
||||
isRateLimited,
|
||||
isRecoveryLimited,
|
||||
issueToken,
|
||||
login,
|
||||
newAccessKey,
|
||||
recordAttempt,
|
||||
recordRecoveryAttempt,
|
||||
recoveryLimitMax,
|
||||
recoveryLimitWindowMs,
|
||||
revokeToken,
|
||||
verifyToken,
|
||||
type PublicUser,
|
||||
} from "./auth.ts";
|
||||
import {
|
||||
createOllamaNode,
|
||||
DEFAULT_NODE_ID,
|
||||
deleteOllamaNode,
|
||||
effectiveOllamaNodes,
|
||||
getOllamaDefaultWeight,
|
||||
getOllamaNode,
|
||||
getOllamaTimeoutMs,
|
||||
listOllamaNodes,
|
||||
nodeStats,
|
||||
resolveOllamaHost,
|
||||
sanitizeNodeHost,
|
||||
sanitizeNodeName,
|
||||
sanitizeWeight,
|
||||
updateOllamaNode,
|
||||
type StatsWindow,
|
||||
} from "./ollamaNodes.ts";
|
||||
import {
|
||||
addMessage,
|
||||
consumeReset,
|
||||
@@ -123,6 +144,7 @@ import {
|
||||
} from "./cdn.ts";
|
||||
import { OCR_MAX_BYTES, ocrAllowed, recordOcrJob } from "./tools/ocr.ts";
|
||||
import { tools } from "./tools/registry.ts";
|
||||
import { initOllamaNodes } from "./ollamaNodes.ts";
|
||||
|
||||
assertConfig();
|
||||
await initDb();
|
||||
@@ -134,6 +156,7 @@ await initAccessTables();
|
||||
await initReportsTables();
|
||||
await initModelUsageTables();
|
||||
await initTotpTables();
|
||||
await initOllamaNodes();
|
||||
|
||||
function json(data: unknown, status = 200): Response {
|
||||
return new Response(JSON.stringify(data), {
|
||||
@@ -142,6 +165,64 @@ function json(data: unknown, status = 200): Response {
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Live probe of one Ollama host: version, latency, on-disk models and
|
||||
* running models (/api/ps — the only load signal Ollama exposes, with
|
||||
* per-model VRAM usage). Always resolves (reachability is data).
|
||||
*/
|
||||
async function probeOllamaHost(host: string): Promise<{
|
||||
reachable: boolean;
|
||||
host: string;
|
||||
version: string | null;
|
||||
latencyMs: number | null;
|
||||
models: { name: string; size: number; modifiedAt: string | null }[];
|
||||
running: { name: string; sizeVram: number; size: number; expiresAt: string | null }[];
|
||||
error: string | null;
|
||||
}> {
|
||||
const t0 = Date.now();
|
||||
try {
|
||||
const verRes = await fetch(`${host}/api/version`, { signal: AbortSignal.timeout(8000) });
|
||||
if (!verRes.ok) {
|
||||
return { reachable: false, host, version: null, latencyMs: Date.now() - t0, models: [], running: [], error: `ollama /api/version failed: ${verRes.status}` };
|
||||
}
|
||||
const ver = (await verRes.json().catch(() => ({}))) as { version?: unknown };
|
||||
const tagsRes = await fetch(`${host}/api/tags`, { signal: AbortSignal.timeout(8000) });
|
||||
const tags = (await tagsRes.json().catch(() => ({}))) as { models?: { name?: unknown; size?: unknown; modified_at?: unknown }[] };
|
||||
const models = tagsRes.ok && Array.isArray(tags.models)
|
||||
? tags.models
|
||||
.filter((m) => typeof m.name === "string")
|
||||
.map((m) => ({
|
||||
name: m.name as string,
|
||||
size: typeof m.size === "number" ? m.size : 0,
|
||||
modifiedAt: typeof m.modified_at === "string" ? (m.modified_at as string) : null,
|
||||
}))
|
||||
: [];
|
||||
const psRes = await fetch(`${host}/api/ps`, { signal: AbortSignal.timeout(8000) });
|
||||
const ps = (await psRes.json().catch(() => ({}))) as { models?: { name?: unknown; size_vram?: unknown; size?: unknown; expires_at?: unknown }[] };
|
||||
const running = psRes.ok && Array.isArray(ps.models)
|
||||
? ps.models
|
||||
.filter((m) => typeof m.name === "string")
|
||||
.map((m) => ({
|
||||
name: m.name as string,
|
||||
sizeVram: typeof m.size_vram === "number" ? m.size_vram : 0,
|
||||
size: typeof m.size === "number" ? m.size : 0,
|
||||
expiresAt: typeof m.expires_at === "string" ? (m.expires_at as string) : null,
|
||||
}))
|
||||
: [];
|
||||
return {
|
||||
reachable: true,
|
||||
host,
|
||||
version: typeof ver.version === "string" ? ver.version : null,
|
||||
latencyMs: Date.now() - t0,
|
||||
models,
|
||||
running,
|
||||
error: null,
|
||||
};
|
||||
} catch (e) {
|
||||
return { reachable: false, host, version: null, latencyMs: Date.now() - t0, models: [], running: [], error: (e as Error).message };
|
||||
}
|
||||
}
|
||||
|
||||
function clientIp(req: Request, server: { requestIP?: (r: Request) => { address: string } | null }): string {
|
||||
try {
|
||||
const info = server.requestIP?.(req);
|
||||
@@ -487,6 +568,16 @@ const server = Bun.serve<{ ticketId: string | null; isAdmin: boolean }>({
|
||||
if (path === "/api/auth/recover" && req.method === "POST") {
|
||||
const ip = clientIp(req, server);
|
||||
if (isRateLimited(ip)) return json({ error: "too many attempts, try again later" }, 429);
|
||||
// Dedicated recovery envelope (admin-configurable, stricter than
|
||||
// login): every hit counts, successes included (anonymous by design).
|
||||
const rl = await isRecoveryLimited(ip);
|
||||
if (rl.limited) {
|
||||
return new Response(
|
||||
JSON.stringify({ error: "too many recovery attempts, try again later", retryAfterSec: rl.retryAfterSec }),
|
||||
{ status: 429, headers: { "Content-Type": "application/json", "Cache-Control": "no-store", "Retry-After": String(rl.retryAfterSec) } },
|
||||
);
|
||||
}
|
||||
recordRecoveryAttempt(ip);
|
||||
const { ok, body } = await readJson(req);
|
||||
// Never enumerate accounts: always answer ok.
|
||||
if (ok && typeof body.username === "string") {
|
||||
@@ -507,6 +598,15 @@ const server = Bun.serve<{ ticketId: string | null; isAdmin: boolean }>({
|
||||
|
||||
const resetMatch = path.match(/^\/api\/auth\/reset\/([A-Za-z0-9_-]{6,80})$/);
|
||||
if (resetMatch) {
|
||||
const ip = clientIp(req, server);
|
||||
const rl = await isRecoveryLimited(ip);
|
||||
if (rl.limited) {
|
||||
return new Response(
|
||||
JSON.stringify({ error: "too many recovery attempts, try again later", retryAfterSec: rl.retryAfterSec }),
|
||||
{ status: 429, headers: { "Content-Type": "application/json", "Cache-Control": "no-store", "Retry-After": String(rl.retryAfterSec) } },
|
||||
);
|
||||
}
|
||||
recordRecoveryAttempt(ip);
|
||||
const tokenHash = createHash("sha256").update(resetMatch[1]!).digest("hex");
|
||||
if (req.method === "GET") {
|
||||
const r = await peekReset(tokenHash);
|
||||
@@ -866,6 +966,10 @@ const server = Bun.serve<{ ticketId: string | null; isAdmin: boolean }>({
|
||||
ocrEnabled: (await getSetting("tools_ocr_enabled", "1")) === "1",
|
||||
reportsEnabled: await isReportsEnabled(),
|
||||
usernameChangeEnabled: await isUsernameChangeEnabled(),
|
||||
ollamaTimeoutS: Math.round(await getOllamaTimeoutMs() / 1000),
|
||||
ollamaDefaultWeight: await getOllamaDefaultWeight(),
|
||||
recoveryLimitMax: await recoveryLimitMax(),
|
||||
recoveryLimitWindowMin: Math.round(await recoveryLimitWindowMs() / 60_000),
|
||||
},
|
||||
});
|
||||
}
|
||||
@@ -901,6 +1005,30 @@ const server = Bun.serve<{ ticketId: string | null; isAdmin: boolean }>({
|
||||
if (typeof body.usernameChangeEnabled !== "boolean") return json({ error: "usernameChangeEnabled must be boolean" }, 400);
|
||||
await setSetting("username_change_enabled", body.usernameChangeEnabled ? "1" : "0");
|
||||
}
|
||||
if (body.ollamaTimeoutS !== undefined) {
|
||||
if (!Number.isInteger(body.ollamaTimeoutS) || (body.ollamaTimeoutS as number) < 5 || (body.ollamaTimeoutS as number) > 3600) {
|
||||
return json({ error: "ollamaTimeoutS must be an integer 5-3600 (seconds)" }, 400);
|
||||
}
|
||||
await setSetting("ollama_timeout_s", String(body.ollamaTimeoutS));
|
||||
}
|
||||
if (body.ollamaDefaultWeight !== undefined) {
|
||||
if (!Number.isInteger(body.ollamaDefaultWeight) || (body.ollamaDefaultWeight as number) < 0 || (body.ollamaDefaultWeight as number) > 1000) {
|
||||
return json({ error: "ollamaDefaultWeight must be an integer 0-1000 (0 = never use the default node)" }, 400);
|
||||
}
|
||||
await setSetting("ollama_default_weight", String(body.ollamaDefaultWeight));
|
||||
}
|
||||
if (body.recoveryLimitMax !== undefined) {
|
||||
if (!Number.isInteger(body.recoveryLimitMax) || (body.recoveryLimitMax as number) < 1 || (body.recoveryLimitMax as number) > 100) {
|
||||
return json({ error: "recoveryLimitMax must be an integer 1-100" }, 400);
|
||||
}
|
||||
await setSetting("recovery_limit_max", String(body.recoveryLimitMax));
|
||||
}
|
||||
if (body.recoveryLimitWindowMin !== undefined) {
|
||||
if (!Number.isInteger(body.recoveryLimitWindowMin) || (body.recoveryLimitWindowMin as number) < 1 || (body.recoveryLimitWindowMin as number) > 1440) {
|
||||
return json({ error: "recoveryLimitWindowMin must be an integer 1-1440 (minutes)" }, 400);
|
||||
}
|
||||
await setSetting("recovery_limit_window_min", String(body.recoveryLimitWindowMin));
|
||||
}
|
||||
return json({
|
||||
settings: {
|
||||
registrationEnabled: await isRegistrationEnabled(),
|
||||
@@ -908,6 +1036,10 @@ const server = Bun.serve<{ ticketId: string | null; isAdmin: boolean }>({
|
||||
ocrEnabled: (await getSetting("tools_ocr_enabled", "1")) === "1",
|
||||
reportsEnabled: await isReportsEnabled(),
|
||||
usernameChangeEnabled: await isUsernameChangeEnabled(),
|
||||
ollamaTimeoutS: Math.round(await getOllamaTimeoutMs() / 1000),
|
||||
ollamaDefaultWeight: await getOllamaDefaultWeight(),
|
||||
recoveryLimitMax: await recoveryLimitMax(),
|
||||
recoveryLimitWindowMin: Math.round(await recoveryLimitWindowMs() / 60_000),
|
||||
},
|
||||
});
|
||||
}
|
||||
@@ -965,6 +1097,16 @@ const server = Bun.serve<{ ticketId: string | null; isAdmin: boolean }>({
|
||||
const name = typeof body.name === "string" ? body.name.trim() : "";
|
||||
if (!/^[a-z0-9._:\/-]{1,100}$/i.test(name)) return json({ error: "invalid model name" }, 400);
|
||||
if (!ollamaDriver.enabled) return json({ error: "ollama driver is disabled" }, 502);
|
||||
// Target node for the pull (library tags AND hf.co/... references).
|
||||
let pullHost = ollamaDriver.host;
|
||||
if (body.nodeId !== undefined) {
|
||||
if (typeof body.nodeId !== "string") return json({ error: "nodeId must be a string" }, 400);
|
||||
try {
|
||||
pullHost = (await resolveOllamaHost(body.nodeId === DEFAULT_NODE_ID ? undefined : body.nodeId)).host;
|
||||
} catch {
|
||||
return json({ error: "unknown ollama node" }, 404);
|
||||
}
|
||||
}
|
||||
const abortUpstream = new AbortController();
|
||||
const onClientAbort = () => abortUpstream.abort();
|
||||
req.signal.addEventListener("abort", onClientAbort);
|
||||
@@ -983,7 +1125,7 @@ const server = Bun.serve<{ ticketId: string | null; isAdmin: boolean }>({
|
||||
}
|
||||
};
|
||||
try {
|
||||
const res = await fetch(`${ollamaDriver.host}/api/pull`, {
|
||||
const res = await fetch(`${pullHost}/api/pull`, {
|
||||
method: "POST",
|
||||
headers: { "Content-Type": "application/json" },
|
||||
body: JSON.stringify({ model: name, stream: true }),
|
||||
@@ -1038,8 +1180,17 @@ const server = Bun.serve<{ ticketId: string | null; isAdmin: boolean }>({
|
||||
if (ollamaDeleteMatch && req.method === "DELETE") {
|
||||
if (!admin) return json({ error: "forbidden" }, 403);
|
||||
if (!ollamaDriver.enabled) return json({ error: "ollama driver is disabled" }, 502);
|
||||
const nodeParam = url.searchParams.get("node");
|
||||
let deleteHost = ollamaDriver.host;
|
||||
if (nodeParam) {
|
||||
try {
|
||||
deleteHost = (await resolveOllamaHost(nodeParam === DEFAULT_NODE_ID ? undefined : nodeParam)).host;
|
||||
} catch {
|
||||
return json({ error: "unknown ollama node" }, 404);
|
||||
}
|
||||
}
|
||||
try {
|
||||
const res = await fetch(`${ollamaDriver.host}/api/delete`, {
|
||||
const res = await fetch(`${deleteHost}/api/delete`, {
|
||||
method: "DELETE",
|
||||
headers: { "Content-Type": "application/json" },
|
||||
body: JSON.stringify({ model: ollamaDeleteMatch[1] }),
|
||||
@@ -1168,6 +1319,117 @@ const server = Bun.serve<{ ticketId: string | null; isAdmin: boolean }>({
|
||||
}
|
||||
}
|
||||
|
||||
// --- admin: ollama nodes (multi-host pool, firewall-like weights) ---
|
||||
if (path === "/api/admin/ollama/nodes" && req.method === "GET") {
|
||||
if (!admin) return json({ error: "forbidden" }, 403);
|
||||
return json({ nodes: await listOllamaNodes(), effective: await effectiveOllamaNodes() });
|
||||
}
|
||||
|
||||
if (path === "/api/admin/ollama/nodes" && req.method === "POST") {
|
||||
if (!admin) return json({ error: "forbidden" }, 403);
|
||||
const { ok, body } = await readJson(req);
|
||||
if (!ok) return json({ error: "invalid JSON" }, 400);
|
||||
const name = sanitizeNodeName(body.name);
|
||||
const host = sanitizeNodeHost(body.host);
|
||||
const weight = sanitizeWeight(body.weight ?? 100);
|
||||
if (!name) return json({ error: "name: 1-60 chars" }, 400);
|
||||
if (!host) return json({ error: "host: http(s) base URL without path (e.g. http://10.0.0.5:11434)" }, 400);
|
||||
if (weight === null) return json({ error: "weight: integer 0-1000" }, 400);
|
||||
if (body.enabled !== undefined && typeof body.enabled !== "boolean") {
|
||||
return json({ error: "enabled must be boolean" }, 400);
|
||||
}
|
||||
try {
|
||||
const node = await createOllamaNode({ name, host, weight, enabled: body.enabled ?? true });
|
||||
registry.invalidateModelsCache();
|
||||
return json({ node }, 201);
|
||||
} catch (e) {
|
||||
if ((e as Error).message === "HOST_TAKEN") return json({ error: "this host is already registered" }, 409);
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
|
||||
const nodeMatch = path.match(/^\/api\/admin\/ollama\/nodes\/([A-Za-z0-9_-]{1,64})$/);
|
||||
if (nodeMatch && !path.endsWith("/status") && !path.endsWith("/stats")) {
|
||||
if (!admin) return json({ error: "forbidden" }, 403);
|
||||
const id = nodeMatch[1]!;
|
||||
if (id === DEFAULT_NODE_ID) return json({ error: "the default node comes from OLLAMA_HOST and cannot be edited here" }, 400);
|
||||
if (req.method === "DELETE") {
|
||||
if (!await deleteOllamaNode(id)) return json({ error: "node not found" }, 404);
|
||||
registry.invalidateModelsCache();
|
||||
return json({ ok: true });
|
||||
}
|
||||
if (req.method === "PATCH") {
|
||||
const { ok, body } = await readJson(req);
|
||||
if (!ok) return json({ error: "invalid JSON" }, 400);
|
||||
const patch: { name?: string; host?: string; weight?: number; enabled?: boolean } = {};
|
||||
if (body.name !== undefined) {
|
||||
const name = sanitizeNodeName(body.name);
|
||||
if (!name) return json({ error: "name: 1-60 chars" }, 400);
|
||||
patch.name = name;
|
||||
}
|
||||
if (body.host !== undefined) {
|
||||
const host = sanitizeNodeHost(body.host);
|
||||
if (!host) return json({ error: "host: http(s) base URL without path" }, 400);
|
||||
patch.host = host;
|
||||
}
|
||||
if (body.weight !== undefined) {
|
||||
const weight = sanitizeWeight(body.weight);
|
||||
if (weight === null) return json({ error: "weight: integer 0-1000" }, 400);
|
||||
patch.weight = weight;
|
||||
}
|
||||
if (body.enabled !== undefined) {
|
||||
if (typeof body.enabled !== "boolean") return json({ error: "enabled must be boolean" }, 400);
|
||||
patch.enabled = body.enabled;
|
||||
}
|
||||
try {
|
||||
const updated = await updateOllamaNode(id, patch);
|
||||
if (!updated) return json({ error: "node not found" }, 404);
|
||||
registry.invalidateModelsCache();
|
||||
return json({ node: updated });
|
||||
} catch (e) {
|
||||
if ((e as Error).message === "HOST_TAKEN") return json({ error: "this host is already registered" }, 409);
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
return json({ error: "not found" }, 404);
|
||||
}
|
||||
|
||||
// Per-node live probe: version, latency, on-disk models, running models
|
||||
// (/api/ps — the only load signal Ollama exposes, with VRAM usage).
|
||||
const nodeStatusMatch = path.match(/^\/api\/admin\/ollama\/nodes\/([A-Za-z0-9_-]{1,64})\/status$/);
|
||||
if (nodeStatusMatch && req.method === "GET") {
|
||||
if (!admin) return json({ error: "forbidden" }, 403);
|
||||
let host: string;
|
||||
try {
|
||||
host = (await resolveOllamaHost(nodeStatusMatch[1] === DEFAULT_NODE_ID ? undefined : nodeStatusMatch[1])).host;
|
||||
} catch {
|
||||
return json({ error: "unknown ollama node" }, 404);
|
||||
}
|
||||
return json(await probeOllamaHost(host));
|
||||
}
|
||||
|
||||
// Probe an arbitrary (not yet registered) host from the node form.
|
||||
if (path === "/api/admin/ollama/probe" && req.method === "GET") {
|
||||
if (!admin) return json({ error: "forbidden" }, 403);
|
||||
const host = sanitizeNodeHost(url.searchParams.get("host"));
|
||||
if (!host) return json({ error: "host: http(s) base URL without path" }, 400);
|
||||
return json(await probeOllamaHost(host));
|
||||
}
|
||||
|
||||
// Per-node generation telemetry aggregated over a window.
|
||||
const nodeStatsMatch = path.match(/^\/api\/admin\/ollama\/nodes\/([A-Za-z0-9_-]{1,64})\/stats$/);
|
||||
if (nodeStatsMatch && req.method === "GET") {
|
||||
if (!admin) return json({ error: "forbidden" }, 403);
|
||||
const raw = (url.searchParams.get("window") ?? "24h").toLowerCase();
|
||||
const win: StatsWindow = raw === "1h" || raw === "6h" || raw === "24h" || raw === "7d" || raw === "30d" ? raw : "24h";
|
||||
try {
|
||||
await resolveOllamaHost(nodeStatsMatch[1] === DEFAULT_NODE_ID ? undefined : nodeStatsMatch[1]);
|
||||
} catch {
|
||||
return json({ error: "unknown ollama node" }, 404);
|
||||
}
|
||||
return json({ stats: await nodeStats(nodeStatsMatch[1]!, win) });
|
||||
}
|
||||
|
||||
// --- admin: access-request wishlist triage ---
|
||||
if (path === "/api/admin/access" && req.method === "GET") {
|
||||
if (!admin) return json({ error: "forbidden" }, 403);
|
||||
|
||||
@@ -0,0 +1,337 @@
|
||||
import { randomBytes } from "node:crypto";
|
||||
import { db, dbKind, getSetting } from "./db.ts";
|
||||
import { config } from "./config.ts";
|
||||
|
||||
/**
|
||||
* Multi-node Ollama pool + generation telemetry.
|
||||
*
|
||||
* - `ollama_nodes`: admin-managed extra Ollama hosts. The env-configured
|
||||
* host (`OLLAMA_HOST`) always exists as the implicit "default" node; its
|
||||
* traffic share is `ollama_default_weight` (0 = never use it).
|
||||
* - Routing is firewall-like: nodes are ordered by weight (highest first)
|
||||
* and each generation picks a node by weighted random among the enabled
|
||||
* ones, failing over to the next heaviest node on error/timeout.
|
||||
* - `ollama_stats`: one row per generation attempt (node, ok, latency,
|
||||
* token counts from Ollama's final chunk) powering the per-node detail
|
||||
* charts. Rows older than 35 days are pruned on write.
|
||||
*
|
||||
* NOTE on host metrics: Ollama exposes no CPU/RAM API — only per-model
|
||||
* VRAM via /api/ps (surfaced in the node detail view). Load-over-time
|
||||
* charts come from app-side telemetry (requests, errors, latency,
|
||||
* tokens/s), which is also what drives routing decisions.
|
||||
*/
|
||||
|
||||
export interface OllamaNode {
|
||||
id: string;
|
||||
name: string;
|
||||
host: string;
|
||||
weight: number;
|
||||
enabled: boolean;
|
||||
created_at: number;
|
||||
updated_at: number;
|
||||
}
|
||||
|
||||
export interface EffectiveNode {
|
||||
/** "default" for the env host, otherwise the row id. */
|
||||
id: string;
|
||||
name: string;
|
||||
host: string;
|
||||
weight: number;
|
||||
}
|
||||
|
||||
export const DEFAULT_NODE_ID = "default";
|
||||
const STATS_RETENTION_MS = 35 * 86400_000;
|
||||
|
||||
export async function initOllamaNodes(): Promise<void> {
|
||||
const autoId = dbKind() === "postgres" ? "BIGSERIAL PRIMARY KEY" : "INTEGER PRIMARY KEY AUTOINCREMENT";
|
||||
await db().exec(`CREATE TABLE IF NOT EXISTS ollama_nodes (
|
||||
id TEXT PRIMARY KEY,
|
||||
name TEXT NOT NULL,
|
||||
host TEXT NOT NULL UNIQUE,
|
||||
weight INTEGER NOT NULL DEFAULT 100,
|
||||
enabled INTEGER NOT NULL DEFAULT 1,
|
||||
created_at BIGINT NOT NULL,
|
||||
updated_at BIGINT NOT NULL
|
||||
)`);
|
||||
await db().exec(`CREATE TABLE IF NOT EXISTS ollama_stats (
|
||||
id ${autoId},
|
||||
node_id TEXT NOT NULL,
|
||||
ts BIGINT NOT NULL,
|
||||
ok INTEGER NOT NULL,
|
||||
latency_ms INTEGER NOT NULL DEFAULT 0,
|
||||
prompt_tokens INTEGER NOT NULL DEFAULT 0,
|
||||
eval_tokens INTEGER NOT NULL DEFAULT 0,
|
||||
model TEXT NOT NULL DEFAULT ''
|
||||
)`);
|
||||
await db().exec(`CREATE INDEX IF NOT EXISTS idx_ollama_stats_node_ts ON ollama_stats(node_id, ts)`);
|
||||
}
|
||||
|
||||
function rowToNode(r: Record<string, number | string>): OllamaNode {
|
||||
return {
|
||||
id: String(r.id),
|
||||
name: String(r.name),
|
||||
host: String(r.host),
|
||||
weight: Number(r.weight),
|
||||
enabled: Number(r.enabled) === 1,
|
||||
created_at: Number(r.created_at),
|
||||
updated_at: Number(r.updated_at),
|
||||
};
|
||||
}
|
||||
|
||||
export async function listOllamaNodes(): Promise<OllamaNode[]> {
|
||||
const rows = await db().all<Record<string, number | string>>(
|
||||
"SELECT * FROM ollama_nodes ORDER BY weight DESC, created_at ASC",
|
||||
);
|
||||
return rows.map(rowToNode);
|
||||
}
|
||||
|
||||
export async function getOllamaNode(id: string): Promise<OllamaNode | null> {
|
||||
const row = await db().get<Record<string, number | string>>("SELECT * FROM ollama_nodes WHERE id = ?", id);
|
||||
return row ? rowToNode(row) : null;
|
||||
}
|
||||
|
||||
function newNodeId(): string {
|
||||
return "node_" + randomBytes(9).toString("base64url");
|
||||
}
|
||||
|
||||
/** Normalize + validate an Ollama base URL (scheme + host, no path). */
|
||||
export function sanitizeNodeHost(raw: unknown): string | null {
|
||||
if (typeof raw !== "string") return null;
|
||||
const t = raw.trim().replace(/\/+$/, "");
|
||||
if (t.length < 10 || t.length > 200) return null;
|
||||
let u: URL;
|
||||
try {
|
||||
u = new URL(t);
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
if (u.protocol !== "http:" && u.protocol !== "https:") return null;
|
||||
if (!u.hostname || u.username || u.password) return null;
|
||||
// Drop default ports and any path/query/fragment for canonical form.
|
||||
const port = u.port && !((u.protocol === "http:" && u.port === "80") || (u.protocol === "https:" && u.port === "443"))
|
||||
? `:${u.port}`
|
||||
: "";
|
||||
if (u.pathname !== "/" && u.pathname !== "") return null;
|
||||
return `${u.protocol}//${u.hostname}${port}`;
|
||||
}
|
||||
|
||||
export function sanitizeNodeName(raw: unknown): string | null {
|
||||
if (typeof raw !== "string") return null;
|
||||
const t = raw.trim();
|
||||
if (t.length === 0 || t.length > 60) return null;
|
||||
return t;
|
||||
}
|
||||
|
||||
export function sanitizeWeight(raw: unknown): number | null {
|
||||
if (typeof raw !== "number" || !Number.isInteger(raw) || raw < 0 || raw > 1000) return null;
|
||||
return raw;
|
||||
}
|
||||
|
||||
export async function createOllamaNode(opts: { name: string; host: string; weight: number; enabled: boolean }): Promise<OllamaNode> {
|
||||
const now = Date.now();
|
||||
const row: OllamaNode = { id: newNodeId(), ...opts, created_at: now, updated_at: now };
|
||||
try {
|
||||
await db().run(
|
||||
"INSERT INTO ollama_nodes (id, name, host, weight, enabled, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
|
||||
row.id, row.name, row.host, row.weight, row.enabled ? 1 : 0, now, now,
|
||||
);
|
||||
} catch (e) {
|
||||
if (/unique|duplicate/i.test((e as Error).message)) throw new Error("HOST_TAKEN");
|
||||
throw e;
|
||||
}
|
||||
return row;
|
||||
}
|
||||
|
||||
export async function updateOllamaNode(
|
||||
id: string,
|
||||
patch: { name?: string; host?: string; weight?: number; enabled?: boolean },
|
||||
): Promise<OllamaNode | null> {
|
||||
const cur = await getOllamaNode(id);
|
||||
if (!cur) return null;
|
||||
const next: OllamaNode = {
|
||||
...cur,
|
||||
name: patch.name ?? cur.name,
|
||||
host: patch.host ?? cur.host,
|
||||
weight: patch.weight ?? cur.weight,
|
||||
enabled: patch.enabled ?? cur.enabled,
|
||||
updated_at: Date.now(),
|
||||
};
|
||||
try {
|
||||
await db().run(
|
||||
"UPDATE ollama_nodes SET name = ?, host = ?, weight = ?, enabled = ?, updated_at = ? WHERE id = ?",
|
||||
next.name, next.host, next.weight, next.enabled ? 1 : 0, next.updated_at, id,
|
||||
);
|
||||
} catch (e) {
|
||||
if (/unique|duplicate/i.test((e as Error).message)) throw new Error("HOST_TAKEN");
|
||||
throw e;
|
||||
}
|
||||
return next;
|
||||
}
|
||||
|
||||
export async function deleteOllamaNode(id: string): Promise<boolean> {
|
||||
const r = await db().run("DELETE FROM ollama_nodes WHERE id = ?", id);
|
||||
return r.changes > 0;
|
||||
}
|
||||
|
||||
// --- effective pool (env default + custom rows) ---
|
||||
|
||||
function clampInt(raw: string, min: number, max: number, fallback: number): number {
|
||||
const v = Number.parseInt(raw, 10);
|
||||
if (!Number.isInteger(v)) return fallback;
|
||||
return Math.min(max, Math.max(min, v));
|
||||
}
|
||||
|
||||
/** Generation timeout for Ollama chat requests (admin setting, seconds). */
|
||||
export async function getOllamaTimeoutMs(): Promise<number> {
|
||||
return clampInt(await getSetting("ollama_timeout_s", "300"), 5, 3600, 300) * 1000;
|
||||
}
|
||||
|
||||
/** Traffic weight of the implicit env node (0 = never use it). */
|
||||
export async function getOllamaDefaultWeight(): Promise<number> {
|
||||
return clampInt(await getSetting("ollama_default_weight", "100"), 0, 1000, 100);
|
||||
}
|
||||
|
||||
/** All routable nodes, heaviest first (default env node included). */
|
||||
export async function effectiveOllamaNodes(): Promise<EffectiveNode[]> {
|
||||
const out: EffectiveNode[] = [];
|
||||
if (config.ollamaEnabled) {
|
||||
const w = await getOllamaDefaultWeight();
|
||||
if (w > 0) {
|
||||
out.push({ id: DEFAULT_NODE_ID, name: "default", host: config.ollamaHost.replace(/\/$/, ""), weight: w });
|
||||
}
|
||||
}
|
||||
for (const n of await listOllamaNodes()) {
|
||||
if (n.enabled && n.weight > 0) out.push({ id: n.id, name: n.name, host: n.host, weight: n.weight });
|
||||
}
|
||||
out.sort((a, b) => b.weight - a.weight);
|
||||
return out;
|
||||
}
|
||||
|
||||
/**
|
||||
* Pick the node for one generation (weighted random) + the failover order
|
||||
* (picked node first, then heaviest-first). Throws when nothing is routable.
|
||||
*/
|
||||
export async function pickOllamaRoute(): Promise<{ first: EffectiveNode; order: EffectiveNode[] }> {
|
||||
const nodes = await effectiveOllamaNodes();
|
||||
if (nodes.length === 0) throw new Error('no ollama node available (all disabled or weight 0)');
|
||||
const total = nodes.reduce((s, n) => s + n.weight, 0);
|
||||
let r = Math.random() * total;
|
||||
let first = nodes[0]!;
|
||||
for (const n of nodes) {
|
||||
r -= n.weight;
|
||||
if (r <= 0) {
|
||||
first = n;
|
||||
break;
|
||||
}
|
||||
}
|
||||
const order = [first, ...nodes.filter((n) => n.id !== first.id)];
|
||||
return { first, order };
|
||||
}
|
||||
|
||||
/** Resolve an admin-selected node id (pull/delete/status) to its host. */
|
||||
export async function resolveOllamaHost(nodeId: string | undefined): Promise<EffectiveNode> {
|
||||
if (!nodeId || nodeId === DEFAULT_NODE_ID) {
|
||||
return { id: DEFAULT_NODE_ID, name: "default", host: config.ollamaHost.replace(/\/$/, ""), weight: 0 };
|
||||
}
|
||||
const n = await getOllamaNode(nodeId);
|
||||
if (!n) throw new Error("unknown ollama node");
|
||||
return { id: n.id, name: n.name, host: n.host, weight: n.weight };
|
||||
}
|
||||
|
||||
// --- telemetry ---
|
||||
|
||||
export async function recordOllamaStat(s: {
|
||||
node_id: string;
|
||||
ok: boolean;
|
||||
latency_ms: number;
|
||||
prompt_tokens: number;
|
||||
eval_tokens: number;
|
||||
model: string;
|
||||
}): Promise<void> {
|
||||
const d = db();
|
||||
await d.run(
|
||||
"INSERT INTO ollama_stats (node_id, ts, ok, latency_ms, prompt_tokens, eval_tokens, model) VALUES (?, ?, ?, ?, ?, ?, ?)",
|
||||
s.node_id, Date.now(), s.ok ? 1 : 0, Math.max(0, Math.round(s.latency_ms)),
|
||||
Math.max(0, Math.round(s.prompt_tokens)), Math.max(0, Math.round(s.eval_tokens)), s.model.slice(0, 160),
|
||||
);
|
||||
// Best-effort retention (indexed range delete, cheap).
|
||||
try {
|
||||
await d.run("DELETE FROM ollama_stats WHERE ts < ?", Date.now() - STATS_RETENTION_MS);
|
||||
} catch {
|
||||
// stats must never break generations
|
||||
}
|
||||
}
|
||||
|
||||
export type StatsWindow = "1h" | "6h" | "24h" | "7d" | "30d";
|
||||
|
||||
export function statsWindowMs(w: StatsWindow): number {
|
||||
return { "1h": 3600_000, "6h": 6 * 3600_000, "24h": 86400_000, "7d": 7 * 86400_000, "30d": 30 * 86400_000 }[w];
|
||||
}
|
||||
|
||||
export interface StatsBucket {
|
||||
t: number;
|
||||
req: number;
|
||||
err: number;
|
||||
avgMs: number | null;
|
||||
promptTok: number;
|
||||
evalTok: number;
|
||||
}
|
||||
|
||||
export interface NodeStats {
|
||||
window: StatsWindow;
|
||||
from: number;
|
||||
to: number;
|
||||
totals: { req: number; err: number; avgMs: number | null; promptTok: number; evalTok: number; tokPerSec: number | null };
|
||||
buckets: StatsBucket[];
|
||||
}
|
||||
|
||||
const BUCKETS = 24;
|
||||
|
||||
export async function nodeStats(nodeId: string, window: StatsWindow): Promise<NodeStats> {
|
||||
const to = Date.now();
|
||||
const span = statsWindowMs(window);
|
||||
const from = to - span;
|
||||
const bucketMs = Math.floor(span / BUCKETS);
|
||||
// GROUP BY output alias works on both SQLite and Postgres; AVG may come
|
||||
// back as text (pg numeric) so every aggregate is coerced in JS.
|
||||
const rows = await db().all<Record<string, number | string | null>>(
|
||||
`SELECT CAST((ts - ?) / ? AS INTEGER) AS b, COUNT(*) AS req,
|
||||
COUNT(*) - COALESCE(SUM(ok), 0) AS err,
|
||||
AVG(CASE WHEN ok = 1 THEN latency_ms END) AS avgms,
|
||||
COALESCE(SUM(prompt_tokens), 0) AS pt, COALESCE(SUM(eval_tokens), 0) AS et
|
||||
FROM ollama_stats WHERE node_id = ? AND ts >= ? GROUP BY b ORDER BY b ASC`,
|
||||
from, bucketMs, nodeId, from,
|
||||
);
|
||||
const buckets: StatsBucket[] = Array.from({ length: BUCKETS }, (_, i) => ({
|
||||
t: from + i * bucketMs,
|
||||
req: 0,
|
||||
err: 0,
|
||||
avgMs: null,
|
||||
promptTok: 0,
|
||||
evalTok: 0,
|
||||
}));
|
||||
for (const r of rows) {
|
||||
const b = Number(r.b);
|
||||
if (!Number.isInteger(b) || b < 0 || b >= BUCKETS) continue;
|
||||
const slot = buckets[b]!;
|
||||
slot.req = Number(r.req);
|
||||
slot.err = Number(r.err);
|
||||
slot.avgMs = r.avgms === null ? null : Number(r.avgms);
|
||||
slot.promptTok = Number(r.pt);
|
||||
slot.evalTok = Number(r.et);
|
||||
}
|
||||
const totals = buckets.reduce(
|
||||
(a, b) => ({ req: a.req + b.req, err: a.err + b.err, promptTok: a.promptTok + b.promptTok, evalTok: a.evalTok + b.evalTok }),
|
||||
{ req: 0, err: 0, promptTok: 0, evalTok: 0 },
|
||||
);
|
||||
const latRows = buckets.filter((b) => b.avgMs !== null && b.req - b.err > 0);
|
||||
const latWeight = latRows.reduce((s, b) => s + (b.req - b.err), 0);
|
||||
const avgMs = latWeight > 0
|
||||
? latRows.reduce((s, b) => s + b.avgMs! * (b.req - b.err), 0) / latWeight
|
||||
: null;
|
||||
// Tokens/s over the window's active span (first to last non-empty bucket).
|
||||
const active = buckets.filter((b) => b.req > 0);
|
||||
const activeMs = active.length > 1 ? active[active.length - 1]!.t - active[0]!.t + bucketMs : active.length === 1 ? bucketMs : 0;
|
||||
const tokPerSec = activeMs > 0 ? (totals.promptTok + totals.evalTok) / (activeMs / 1000) : null;
|
||||
return { window, from, to, totals: { ...totals, avgMs, tokPerSec }, buckets };
|
||||
}
|
||||
Reference in New Issue
Block a user