diff --git a/web/logs.test.ts b/web/logs.test.ts new file mode 100644 index 0000000..0e72122 --- /dev/null +++ b/web/logs.test.ts @@ -0,0 +1,106 @@ +import { afterEach, describe, expect, it, mock } from 'bun:test' +import { LOG_CAP, parseLogLine, pushCapped, streamLogs } from './logs.ts' +import { LogStreamError } from './logs.ts' + +describe('解析日志行', () => { + it('形状对的行解出级别和内容', () => { + expect(parseLogLine('{"type":"info","payload":"[TCP] dial example.com"}')).toEqual({ + type: 'info', + payload: '[TCP] dial example.com', + }) + }) + + it('不是 JSON 的脏行返回 null 而不是抛出去', () => { + expect(parseLogLine('not json at all')).toBeNull() + expect(parseLogLine('{"type":"info","payl')).toBeNull() + }) + + it('是 JSON 但形状不对的也跳过', () => { + expect(parseLogLine('null')).toBeNull() + expect(parseLogLine('[1,2]')).toBeNull() + expect(parseLogLine('{"type":"info"}')).toBeNull() + expect(parseLogLine('{"type":"info","payload":42}')).toBeNull() + }) +}) + +describe('成批追加上限', () => { + it('超出上限丢最旧的,留下最新的', () => { + const list = [1, 2, 3] + pushCapped(list, [4, 5, 6], 4) + expect(list).toEqual([3, 4, 5, 6]) + }) + + it('一批比上限还大时只留下限条', () => { + const list: number[] = [] + pushCapped(list, [1, 2, 3, 4, 5], 3) + expect(list).toEqual([3, 4, 5]) + }) +}) + +function streamOf(chunks: string[]): ReadableStream { + const encoder = new TextEncoder() + return new ReadableStream({ + start(controller) { + for (const chunk of chunks) controller.enqueue(encoder.encode(chunk)) + controller.close() + }, + }) +} + +const realFetch = globalThis.fetch + +// bun 的测试环境没有 localStorage,而 streamLogs 要从里面取令牌 +if (typeof globalThis.localStorage === 'undefined') { + ;(globalThis as Record).localStorage = { + getItem: () => '', + setItem: () => {}, + removeItem: () => {}, + } +} + +afterEach(() => { + globalThis.fetch = realFetch +}) + +describe('读日志流', () => { + it('脏行被跳过,不会中断后面的行', async () => { + globalThis.fetch = mock(() => + Promise.resolve( + new Response( + streamOf([ + '{"type":"info","payload":"第一条"}\n', + '这不是 JSON\n', + '{"type":"warning","payload":"第', + '二条"}\n', + ]), + ), + ), + ) as unknown as typeof fetch + + const lines: string[] = [] + for await (const line of streamLogs('info', new AbortController().signal)) { + lines.push(`${line.type}: ${line.payload}`) + } + expect(lines).toEqual(['info: 第一条', 'warning: 第二条']) + }) + + it('接口报错时用服务端给的 message 抛出', async () => { + globalThis.fetch = mock(() => + Promise.resolve( + new Response(JSON.stringify({ message: '内核未连接' }), { status: 502 }), + ), + ) as unknown as typeof fetch + + const run = async () => { + for await (const _ of streamLogs('info', new AbortController().signal)) { + // 不该有产出 + } + } + await expect(run()).rejects.toThrow(new LogStreamError('内核未连接')) + }) + + it('LOG_CAP 是个有限值,无限流不会把内存吃光', () => { + expect(LOG_CAP).toBeGreaterThan(0) + expect(LOG_CAP).toBeLessThanOrEqual(10000) + }) +}) diff --git a/web/logs.ts b/web/logs.ts new file mode 100644 index 0000000..9a6ef74 --- /dev/null +++ b/web/logs.ts @@ -0,0 +1,87 @@ +/** + * 内核日志流。 + * + * 接口是 `GET /api/kernel/logs?level=...`,一条永不结束的流,一行一个 JSON, + * 形如 `{ "type": "info", "payload": "..." }`。所以不能走 api.ts 的 request()—— + * 那个会把整个响应读完,对流等于挂死。这里自己 fetch,按行读用 stream.ts。 + * + * 解析和成批追加放在组件外面,是因为「脏行跳过」「超上限丢最旧的」这两条 + * 是验收里点名要测的行为,埋在 .vue 里测不到。 + */ +import { getToken } from './api' +import { readLines } from './stream' + +/** 内核接受的日志级别。连接建立时定死,中途改不了,所以切换等于重连。 */ +export const LEVELS = ['debug', 'info', 'warning', 'error', 'silent'] as const +export type LogLevel = (typeof LEVELS)[number] + +/** + * 页面上保留的最近条数。日志是无限流,不设上限内存会一直涨, + * 而目标机器只有 1-2G。超出就丢最旧的。 + */ +export const LOG_CAP = 1000 + +export interface LogLine { + type: string + payload: string +} + +/** 解析一行日志。脏行(不是 JSON、形状不对)返回 null,由调用方跳过,不能抛。 */ +export function parseLogLine(line: string): LogLine | null { + let value: unknown + try { + value = JSON.parse(line) + } catch { + return null + } + if (typeof value !== 'object' || value === null || Array.isArray(value)) return null + const { type, payload } = value as Record + if (typeof type !== 'string' || typeof payload !== 'string') return null + return { type, payload } +} + +/** 把一批新条目追加进列表,超出上限从头部丢最旧的。 */ +export function pushCapped(list: T[], items: T[], cap: number): void { + list.push(...items) + if (list.length > cap) list.splice(0, list.length - cap) +} + +export class LogStreamError extends Error { + constructor(message: string) { + super(message) + this.name = 'LogStreamError' + } +} + +/** + * 连上内核的日志流,逐条产出。停止的方式是中止 signal。 + * + * 流正常结束(for await 走完)意味着内核那边断了连接,比如内核重启—— + * 调用方要把它和「主动中止」区分开,前者该告诉用户。 + */ +export async function* streamLogs( + level: LogLevel, + signal: AbortSignal, +): AsyncGenerator { + const res = await fetch(`/api/kernel/logs?level=${level}`, { + headers: { Authorization: `Bearer ${getToken()}` }, + signal, + }) + if (!res.ok) { + // 失败时后端回的是一次性的 { message, hint? },不是流 + let message = `日志接口请求失败(HTTP ${String(res.status)})` + try { + const payload = JSON.parse(await res.text()) as Record + if (typeof payload.message === 'string') message = payload.message + } catch { + // 保留默认消息 + } + throw new LogStreamError(message) + } + if (!res.body) throw new LogStreamError('日志接口没有返回数据流') + + for await (const line of readLines(res.body)) { + const entry = parseLogLine(line) + if (entry) yield entry + } +} diff --git a/web/views/RequestsPage.vue b/web/views/RequestsPage.vue new file mode 100644 index 0000000..0a6efad --- /dev/null +++ b/web/views/RequestsPage.vue @@ -0,0 +1,283 @@ + + + + +