From 2bd09c89e94795e7a77fa59c6f3e6942c3d0292e Mon Sep 17 00:00:00 2001 From: root Date: Tue, 6 Oct 2026 19:34:07 +0800 Subject: [PATCH] =?UTF-8?q?fix(logs):=20=E6=97=A5=E5=BF=97=E6=9F=A5?= =?UTF-8?q?=E8=AF=A2=E5=85=88=E6=8C=89=E5=8E=9F=E5=A7=8B=E5=88=97=E7=AD=9B?= =?UTF-8?q?=E9=80=89=E5=8F=96=E5=89=8D=20n=20=E6=9D=A1=E5=86=8D=E7=AE=97?= =?UTF-8?q?=E6=A0=87=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 带筛选(级别、关键词、类型)时原查询要对全部引擎运行算标题子查询与 search 拼接, 3 万条运行上超时。改为每个来源两段:第一段只用原始列与一次性算好的业务引用筛选并 order + limit,第二段只对取出的行算标题 / 对象标签;关键词先在业务对象小表里查出 命中 id 集合;查询内关掉 JIT。新增 3 万条运行的耗时测试与改写前口径对照测试。 Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Cxwgnn2pbzQ7rXv3rq7fei --- apps/server/src/modules/logs/query.ts | 435 ++++++++++++------ apps/server/test/logs/legacy-query.ts | 232 ++++++++++ .../test/logs/logs-legacy-parity.test.ts | 117 +++++ apps/server/test/logs/logs-volume.test.ts | 105 +++++ apps/server/test/logs/volume-seed.ts | 87 ++++ 5 files changed, 834 insertions(+), 142 deletions(-) create mode 100644 apps/server/test/logs/legacy-query.ts create mode 100644 apps/server/test/logs/logs-legacy-parity.test.ts create mode 100644 apps/server/test/logs/logs-volume.test.ts create mode 100644 apps/server/test/logs/volume-seed.ts diff --git a/apps/server/src/modules/logs/query.ts b/apps/server/src/modules/logs/query.ts index 76bcd321..f683e36c 100644 --- a/apps/server/src/modules/logs/query.ts +++ b/apps/server/src/modules/logs/query.ts @@ -5,7 +5,8 @@ import { redact } from "../../logger.ts"; import { BUSINESS_EXPR } from "../auth-settings/engine-business.ts"; // 日志聚合查询(界面第二轮 V 线):四个来源各自出同一组列,按 (at, id) 倒序各取 limit+1 条,union 后再取前 limit 条。 -// 每个来源单独 order + limit,PG 能沿各自的时间索引倒着扫,不需要把全部来源物化再排序。 +// 每个来源单独 order + limit,PG 能沿各自的时间索引倒着扫,不需要把全部来源物化再排序; +// 筛选只用原始列,标题等昂贵列只对取出的行算(见 SourceBuilder)。 // // 来源与口径(docs/runbook/logs.md): // - engine_run → task:引擎一次运行;时间 = 创建时间;标题 = 业务对象名(没有时为工作流名);对象 = 业务对象 @@ -54,11 +55,73 @@ const SOURCE_TYPES: Record = { const UUID_RE = "^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$"; -function engineSource(schema: string): RawBuilder { +/** 关键词命中的业务对象每类最多取这么多个(只影响「按业务标题搜任务」,标题极泛的词会漏掉更早的对象) */ +export const KEYWORD_OBJECT_CAP = 2000; + +/** 各来源共用的已解析筛选 */ +interface Filter { + types: Set | null; + levels: LogLevel[] | null; + from: Date | null; + to: Date | null; + /** ilike 模式(已转义,带 %) */ + like: string | null; + /** 关键词命中的业务对象 "kind:id"(只给引擎来源用) */ + matched: string[]; + objectType: string | null; + objectId: string | null; + hideRoutine: boolean; + cursor: Cursor | null; +} + +/** + * 每个来源分两段: + * 1. 只用原始列(及一次性算好的业务引用)筛选,`order by 时间 desc, 日志 id desc limit n`,PG 沿时间索引倒着扫、凑够 n 条就停; + * 2. 只对这 n 行算标题、对象标签、extra 等昂贵列。 + * 以前是先构造带派生列(标题子查询、search 拼接)的子查询再在外层筛选,带筛选时要对全部引擎运行算一遍派生列。 + */ +type SourceBuilder = (f: Filter, n: number) => RawBuilder; + +/** 时间范围与游标条件;at、logId 是该来源的时间列与日志 id 表达式 */ +function commonConds(f: Filter, at: RawBuilder, logId: RawBuilder): RawBuilder[] { + const out: RawBuilder[] = []; + if (f.from) out.push(sql`${at} >= ${f.from}`); + if (f.to) out.push(sql`${at} < ${f.to}`); + if (f.cursor) { + const ts = sql`(timestamptz 'epoch' + ${f.cursor.us} * interval '1 microsecond')`; + // 拆成 at <= ts(可走索引)+ 同一时刻再比日志 id,与 (at, id) < (ts, id) 等价 + out.push(sql`${at} <= ${ts} and (${at} < ${ts} or ${logId} < ${f.cursor.id})`); + } + return out; +} + +const andAll = (conds: RawBuilder[]) => (conds.length ? sql.join(conds, sql` and `) : sql`true`); +const levelIn = (expr: RawBuilder, f: Filter) => (f.levels ? [sql`${expr} = any(${f.levels}::text[])`] : []); + +function engineSource(schema: string): SourceBuilder { if (!/^[a-z_][a-z0-9_]*$/.test(schema)) throw new Error(`非法 schema 名 ${schema}`); const t = sql.raw(`"${schema}".workflow_runs`); - const kind = sql`(biz.b->>'kind')`; - const ref = sql`(biz.b->>'refId')`; + const st = sql.raw("workflow_runs.status"); + const logId = sql`('engine_run:' || workflow_runs.id)`; + // 第一段:业务引用 BUSINESS_EXPR 每行只算一次(offset 0 阻止子查询上拉,否则每处引用都会把整个表达式展开重算)。 + // 智能体结果(级别、例行要用)的 lateral 也加 offset 0:保持逐行嵌套循环,凑够 n 条就停; + // 不加时规划器会改成与 agent_run 的哈希连接,要先扫完全部运行。 + const kind1 = sql`(biz.b->>'kind')`; + const ref1 = sql`(biz.b->>'refId')`; + const uuid1 = sql`(case when ${ref1} ~* ${UUID_RE} then ${ref1}::uuid end)`; + const level = sql`(case + when ${st} = 'failed' then 'error' + when ${st} = 'canceled' then 'warn' + when ${st} in ('completed', 'succeeded') then (case ag.status + when 'failed' then 'error' when 'budget_exceeded' then 'warn' when 'cancelled' then 'warn' else 'success' end) + else 'running' end)`; + const routine = sql`(biz.b is null + or coalesce(${kind1} = 'schedule' and ${st} in ('completed', 'succeeded'), false) + or coalesce(ag.trigger in ('event', 'schedule'), false))`; + + // 第二段:p.kind / p.ref 是第一段带出来的列 + const kind = sql.raw("p.kind"); + const ref = sql.raw("p.ref"); const uuidRef = sql`(case when ${ref} ~* ${UUID_RE} then ${ref}::uuid end)`; const objectType = sql`(case ${kind} ${sql.join( Object.entries(BUSINESS_OBJECT_TYPE).map(([k, v]) => sql`when ${sql.lit(k)} then ${sql.lit(v)}`), @@ -77,128 +140,211 @@ function engineSource(schema: string): RawBuilder { when 'decision' then (select d.title from decision d where d.id = ${uuidRef}) when 'media' then (select e.title from media_entity e where e.id = ${uuidRef}) end)`; - const agent = sql`(case when ${kind} = 'agent' then - (select jsonb_build_object('status', a.status, 'trigger', a.trigger) from agent_run a where a.id = ${uuidRef}) end)`; - const st = sql.raw("workflow_runs.status"); - return sql`select - 'engine_run:' || workflow_runs.id as id, - 'task'::text as type, - 'engine_run'::text as source, - workflow_runs.id::text as ref_id, - workflow_runs.created_at as at, - workflow_runs.finished_at as finished_at, - (case when workflow_runs.finished_at is not null and workflow_runs.started_at is not null - then extract(epoch from workflow_runs.finished_at - workflow_runs.started_at) * 1000 end)::float8 as duration_ms, - (case - when ${st} = 'failed' then 'error' - when ${st} = 'canceled' then 'warn' - when ${st} in ('completed', 'succeeded') then (case ag.a->>'status' - when 'failed' then 'error' when 'budget_exceeded' then 'warn' when 'cancelled' then 'warn' else 'success' end) - else 'running' end) as level, - ${st}::text as status, - coalesce(bt.title, workflow_runs.workflow_name) as title, - (case when ${st} = 'failed' then left(coalesce(workflow_runs.error->>'message', workflow_runs.error::text), 500) end) - as message, - ${objectType} as object_type, - ${ref} as object_id, - bt.title as object_label, - null::text as actor_type, - null::text as actor_id, - jsonb_build_object('workflowName', workflow_runs.workflow_name, 'businessKind', ${kind}, - 'businessRefId', ${ref}, 'businessTitle', bt.title, 'outcome', ag.a->>'status', - 'agentTrigger', ag.a->>'trigger', 'attempts', workflow_runs.attempts) as extra, - (biz.b is null - or coalesce(${kind} = 'schedule' and ${st} in ('completed', 'succeeded'), false) - or coalesce((ag.a->>'trigger') in ('event', 'schedule'), false)) as routine, - concat_ws(' ', workflow_runs.workflow_name, workflow_runs.id, bt.title, ${ref}, - left(workflow_runs.error::text, 500)) as search - from ${t} - cross join lateral (select ${BUSINESS_EXPR} as b) biz - cross join lateral (select ${title} as title) bt - cross join lateral (select ${agent} as a) ag - where workflow_runs.namespace_id = 'default'`; + + return (f, n) => { + const conds: RawBuilder[] = [ + sql`workflow_runs.namespace_id = 'default'`, + ...commonConds(f, sql`workflow_runs.created_at`, logId), + ]; + conds.push(...levelIn(level, f)); + if (f.hideRoutine) conds.push(sql`not ${routine}`); + if (f.like) { + // 标题命中 → 业务引用在命中集合里;其余按原始列(工作流名、运行 id、业务对象 id、错误) + conds.push(sql`(${f.matched.length ? sql`(${kind1} || ':' || ${ref1}) = any(${f.matched}::text[]) or ` : sql``} + workflow_runs.workflow_name ilike ${f.like} or workflow_runs.id ilike ${f.like} or ${ref1} ilike ${f.like} + or left(workflow_runs.error::text, 500) ilike ${f.like})`); + } + if (f.objectType) { + const kinds = Object.entries(BUSINESS_OBJECT_TYPE) + .filter(([, v]) => v === f.objectType) + .map(([k]) => k); + conds.push(kinds.length ? sql`${kind1} = any(${kinds}::text[])` : sql`false`); + } + if (f.objectId) conds.push(sql`${ref1} = ${f.objectId}`); + return sql`select + 'engine_run:' || p.id as id, + 'task'::text as type, + 'engine_run'::text as source, + p.id::text as ref_id, + p.created_at as at, + p.finished_at as finished_at, + (case when p.finished_at is not null and p.started_at is not null + then extract(epoch from p.finished_at - p.started_at) * 1000 end)::float8 as duration_ms, + p.level, + p.status::text as status, + coalesce(bt.title, p.workflow_name) as title, + (case when p.status = 'failed' then left(coalesce(p.error->>'message', p.error::text), 500) end) as message, + ${objectType} as object_type, + ${ref} as object_id, + bt.title as object_label, + null::text as actor_type, + null::text as actor_id, + jsonb_build_object('workflowName', p.workflow_name, 'businessKind', ${kind}, + 'businessRefId', ${ref}, 'businessTitle', bt.title, 'outcome', p.agent_status, + 'agentTrigger', p.agent_trigger, 'attempts', p.attempts) as extra + from ( + select workflow_runs.id, workflow_runs.created_at, workflow_runs.started_at, workflow_runs.finished_at, + workflow_runs.status, workflow_runs.workflow_name, workflow_runs.error, workflow_runs.attempts, + ${kind1} as kind, ${ref1} as ref, ag.status as agent_status, ag.trigger as agent_trigger, ${level} as level + from ${t} + cross join lateral (select ${BUSINESS_EXPR} as b offset 0) biz + left join lateral (select a.status::text as status, a.trigger::text as trigger from agent_run a + where ${kind1} = 'agent' and a.id = ${uuid1} offset 0) ag on true + where ${andAll(conds)} + order by workflow_runs.created_at desc, ${logId} desc + limit ${n} + ) p + cross join lateral (select ${title} as title offset 0) bt`; + }; } -const SCHEDULE_SOURCE = sql`select - 'schedule_run:' || r.id::text as id, - 'auto'::text as type, - 'schedule_run'::text as source, - r.id::text as ref_id, - r.started_at as at, - r.finished_at as finished_at, - r.duration_ms::float8 as duration_ms, - (case +// 定时任务运行、智能体后台调查、审计:行数少、列都在本行(定时任务只多一个小表 join),两段合成一条查询: +// 级别 / 类型 / 例行 / 关键词都是本行列上的表达式,直接放进 where,再按时间倒序取 n 条。 + +const scheduleSource: SourceBuilder = (f, n) => { + const logId = sql`('schedule_run:' || r.id::text)`; + const level = sql`(case when r.status = 'failed' or r.agent_status = 'failed' then 'error' when r.agent_status in ('budget_exceeded', 'cancelled') then 'warn' when r.status = 'ok' then 'success' when r.status = 'skipped' then 'info' - else 'running' end) as level, - r.status::text as status, - coalesce(j.name, r.job_id) as title, - coalesce(r.error, r.summary) as message, - 'scheduled_job'::text as object_type, - r.job_id as object_id, - j.name as object_label, - r.trigger::text as actor_type, - null::text as actor_id, - jsonb_build_object('jobId', r.job_id, 'jobKind', j.kind, 'trigger', r.trigger, 'agentRunId', r.agent_run_id, - 'agentStatus', r.agent_status, 'summary', r.summary, 'error', r.error) as extra, - (r.status in ('ok', 'skipped') and coalesce(j.kind::text, 'task') <> 'agent_check' - and r.agent_status is distinct from 'failed') as routine, - concat_ws(' ', r.job_id, j.name, r.summary, r.error, r.agent_run_id::text) as search - from scheduled_job_run r left join scheduled_job j on j.id = r.job_id`; + else 'running' end)`; + const conds = [...commonConds(f, sql`r.started_at`, logId), ...levelIn(level, f)]; + if (f.hideRoutine) { + conds.push(sql`not (r.status in ('ok', 'skipped') and coalesce(j.kind::text, 'task') <> 'agent_check' + and r.agent_status is distinct from 'failed')`); + } + if (f.like) { + conds.push(sql`concat_ws(' ', r.job_id, j.name, r.summary, r.error, r.agent_run_id::text) ilike ${f.like}`); + } + if (f.objectType && f.objectType !== "scheduled_job") conds.push(sql`false`); + if (f.objectId) conds.push(sql`r.job_id = ${f.objectId}`); + return sql`select + ${logId} as id, + 'auto'::text as type, + 'schedule_run'::text as source, + r.id::text as ref_id, + r.started_at as at, + r.finished_at as finished_at, + r.duration_ms::float8 as duration_ms, + ${level} as level, + r.status::text as status, + coalesce(j.name, r.job_id) as title, + coalesce(r.error, r.summary) as message, + 'scheduled_job'::text as object_type, + r.job_id as object_id, + j.name as object_label, + r.trigger::text as actor_type, + null::text as actor_id, + jsonb_build_object('jobId', r.job_id, 'jobKind', j.kind, 'trigger', r.trigger, 'agentRunId', r.agent_run_id, + 'agentStatus', r.agent_status, 'summary', r.summary, 'error', r.error) as extra + from scheduled_job_run r left join scheduled_job j on j.id = r.job_id + where ${andAll(conds)} + order by r.started_at desc, ${logId} desc + limit ${n}`; +}; -const AGENT_SOURCE = sql`select - 'agent_run:' || a.id::text as id, - 'auto'::text as type, - 'agent_run'::text as source, - a.id::text as ref_id, - a.created_at as at, - a.finished_at as finished_at, - (case when a.finished_at is not null and a.started_at is not null - then extract(epoch from a.finished_at - a.started_at) * 1000 end)::float8 as duration_ms, - (case a.status +const agentSource: SourceBuilder = (f, n) => { + const logId = sql`('agent_run:' || a.id::text)`; + const level = sql`(case a.status when 'succeeded' then 'success' when 'failed' then 'error' - when 'cancelled' then 'warn' when 'budget_exceeded' then 'warn' else 'running' end) as level, - a.status::text as status, - a.goal as title, - coalesce(a.summary, a.error->>'message', a.spawn_reason) as message, - 'agent_run'::text as object_type, - a.id::text as object_id, - null::text as object_label, - 'event'::text as actor_type, - a.created_by as actor_id, - jsonb_build_object('trigger', a.trigger, 'mode', a.mode, 'toolCalls', a.tool_call_count, - 'spawnReason', a.spawn_reason, 'model', a.model) as extra, - false as routine, - concat_ws(' ', a.goal, a.summary, a.id::text, left(a.error::text, 500)) as search - from agent_run a where a.trigger = 'event' and a.parent_run_id is null`; + when 'cancelled' then 'warn' when 'budget_exceeded' then 'warn' else 'running' end)`; + const conds = [ + sql`a.trigger = 'event' and a.parent_run_id is null`, + ...commonConds(f, sql`a.created_at`, logId), + ...levelIn(level, f), + ]; + if (f.like) { + conds.push(sql`concat_ws(' ', a.goal, a.summary, a.id::text, left(a.error::text, 500)) ilike ${f.like}`); + } + if (f.objectType && f.objectType !== "agent_run") conds.push(sql`false`); + if (f.objectId) conds.push(sql`a.id::text = ${f.objectId}`); + return sql`select + ${logId} as id, + 'auto'::text as type, + 'agent_run'::text as source, + a.id::text as ref_id, + a.created_at as at, + a.finished_at as finished_at, + (case when a.finished_at is not null and a.started_at is not null + then extract(epoch from a.finished_at - a.started_at) * 1000 end)::float8 as duration_ms, + ${level} as level, + a.status::text as status, + a.goal as title, + coalesce(a.summary, a.error->>'message', a.spawn_reason) as message, + 'agent_run'::text as object_type, + a.id::text as object_id, + null::text as object_label, + 'event'::text as actor_type, + a.created_by as actor_id, + jsonb_build_object('trigger', a.trigger, 'mode', a.mode, 'toolCalls', a.tool_call_count, + 'spawnReason', a.spawn_reason, 'model', a.model) as extra + from agent_run a + where ${andAll(conds)} + order by a.created_at desc, ${logId} desc + limit ${n}`; +}; -const AUDIT_SOURCE = sql`select - 'audit_event:' || e.id::text as id, - (case when e.action like 'admin.%' or e.action like 'api_key.%' then 'auth' else 'audit' end) as type, - 'audit_event'::text as source, - e.id::text as ref_id, - e.at as at, - null::timestamptz as finished_at, - null::float8 as duration_ms, - (case +const auditSource: SourceBuilder = (f, n) => { + const logId = sql`('audit_event:' || e.id::text)`; + const type = sql`(case when e.action like 'admin.%' or e.action like 'api_key.%' then 'auth' else 'audit' end)`; + const level = sql`(case when e.action in ('admin.login_failed', 'capability.denied') then 'warn' when e.action in ('admin.login_success', 'admin.passkey_login', 'admin.setup') then 'success' - else 'info' end) as level, - e.action as status, - e.action as title, - (case when e.action like 'admin.%' then concat_ws(' · ', e.payload->>'username', e.payload->>'ip') - else nullif(concat_ws(' · ', e.payload->>'reason', e.payload->>'error'), '') end) as message, - e.target_type as object_type, - e.target_id as object_id, - null::text as object_label, - e.actor_type as actor_type, - e.actor_id as actor_id, - jsonb_build_object('action', e.action, 'traceId', e.trace_id, 'payload', e.payload) as extra, - false as routine, - concat_ws(' ', e.action, e.target_type, e.target_id, e.actor_id, left(e.payload::text, 2000)) as search - from audit_event e`; + else 'info' end)`; + const conds = [...commonConds(f, sql`e.at`, logId), ...levelIn(level, f)]; + if (f.types) conds.push(sql`${type} = any(${[...f.types]}::text[])`); + if (f.like) { + conds.push(sql`concat_ws(' ', e.action, e.target_type, e.target_id, e.actor_id, left(e.payload::text, 2000)) + ilike ${f.like}`); + } + if (f.objectType) conds.push(sql`e.target_type = ${f.objectType}`); + if (f.objectId) conds.push(sql`e.target_id = ${f.objectId}`); + return sql`select + ${logId} as id, + ${type} as type, + 'audit_event'::text as source, + e.id::text as ref_id, + e.at as at, + null::timestamptz as finished_at, + null::float8 as duration_ms, + ${level} as level, + e.action as status, + e.action as title, + (case when e.action like 'admin.%' then concat_ws(' · ', e.payload->>'username', e.payload->>'ip') + else nullif(concat_ws(' · ', e.payload->>'reason', e.payload->>'error'), '') end) as message, + e.target_type as object_type, + e.target_id as object_id, + null::text as object_label, + e.actor_type as actor_type, + e.actor_id as actor_id, + jsonb_build_object('action', e.action, 'traceId', e.trace_id, 'payload', e.payload) as extra + from audit_event e + where ${andAll(conds)} + order by e.at desc, ${logId} desc + limit ${n}`; +}; + +/** + * 关键词命中的业务对象("kind:id"),口径与引擎运行的标题一致(标题 = 业务对象名);每类最多 KEYWORD_OBJECT_CAP 个。 + * 这些表都不大,先查出 id 集合,引擎来源再按业务引用 ∈ 集合筛选,不用对每条运行算标题。 + */ +async function keywordObjects(db: Kysely, like: string): Promise { + const cap = KEYWORD_OBJECT_CAP; + const r = await sql<{ k: string }>` + (select 'schedule:' || j.id as k from scheduled_job j where j.name ilike ${like} limit ${cap}) + union all (select 'agent:' || a.id from agent_run a where a.goal ilike ${like} limit ${cap}) + union all (select 'download:' || d.id from download_task d where d.name ilike ${like} limit ${cap}) + union all (select 'organization:' || r.id from organization_record r left join media_entity e on e.id = r.media_entity_id + where coalesce(e.title, r.target_path, r.source_path) ilike ${like} limit ${cap}) + union all (select 'subscription:' || s.id from subscription s left join media_entity e on e.id = s.media_entity_id + where coalesce(e.title, s.title_query) ilike ${like} limit ${cap}) + union all (select 'feed:' || f.id from feed_source f where f.name ilike ${like} limit ${cap}) + union all (select 'site:' || s.id from site s where s.name ilike ${like} limit ${cap}) + union all (select 'decision:' || d.id from decision d where d.title ilike ${like} limit ${cap}) + union all (select 'media:' || e.id from media_entity e where e.title ilike ${like} limit ${cap})`.execute(db); + return r.rows.map((x) => x.k); +} type Row = { id: string; @@ -234,6 +380,7 @@ export interface LogQueries { export function logQueries(db: Kysely, opts: { engineSchema?: string } = {}): LogQueries { const engineSchema = opts.engineSchema ?? "openworkflow"; + const engine = engineSource(engineSchema); let engineReady = false; const engineExists = async () => { if (engineReady) return true; @@ -247,35 +394,39 @@ export function logQueries(db: Kysely, opts: { engineSchema?: string } = {}) return { async list(q) { const types = q.type?.length ? new Set(q.type) : null; - const sources: [LogSource, RawBuilder][] = []; const want = (s: LogSource) => !types || SOURCE_TYPES[s].some((t) => types.has(t)); - if (want("engine_run") && (await engineExists())) sources.push(["engine_run", engineSource(engineSchema)]); - if (want("schedule_run")) sources.push(["schedule_run", SCHEDULE_SOURCE]); - if (want("agent_run")) sources.push(["agent_run", AGENT_SOURCE]); - if (want("audit_event")) sources.push(["audit_event", AUDIT_SOURCE]); - if (!sources.length) return { items: [], nextCursor: null }; - - const conds: RawBuilder[] = []; - if (types) conds.push(sql`s.type = any(${[...types]}::text[])`); - if (q.level?.length) conds.push(sql`s.level = any(${q.level}::text[])`); - if (q.from) conds.push(sql`s.at >= ${new Date(q.from)}`); - if (q.to) conds.push(sql`s.at < ${new Date(q.to)}`); - if (q.q) conds.push(sql`s.search ilike ${`%${escapeLike(q.q)}%`}`); - if (q.objectType) conds.push(sql`s.object_type = ${q.objectType}`); - if (q.objectId) conds.push(sql`s.object_id = ${q.objectId}`); - if (q.system !== "1" && !q.objectId) conds.push(sql`not s.routine`); - const cur = decodeCursor(q.cursor); - if (cur) { - conds.push(sql`(s.at, s.id) < (timestamptz 'epoch' + ${cur.us} * interval '1 microsecond', ${cur.id})`); - } - const where = conds.length ? sql`where ${sql.join(conds, sql` and `)}` : sql``; + const withEngine = want("engine_run") && (await engineExists()); + const like = q.q ? `%${escapeLike(q.q)}%` : null; + const f: Filter = { + types, + levels: q.level?.length ? q.level : null, + from: q.from ? new Date(q.from) : null, + to: q.to ? new Date(q.to) : null, + like, + matched: like && withEngine ? await keywordObjects(db, like) : [], + objectType: q.objectType || null, + objectId: q.objectId || null, + hideRoutine: q.system !== "1" && !q.objectId, + cursor: decodeCursor(q.cursor), + }; const n = q.limit + 1; - const parts = sources.map( - ([, src]) => sql`(select s.* from (${src}) s ${where} order by s.at desc, s.id desc limit ${n})`, - ); - const rows = await sql`select u.*, (extract(epoch from u.at) * 1000000)::bigint as cursor_us - from (${sql.join(parts, sql` union all `)}) u - order by u.at desc, u.id desc limit ${n}`.execute(db); + const parts: RawBuilder[] = []; + if (withEngine) parts.push(engine(f, n)); + if (want("schedule_run")) parts.push(scheduleSource(f, n)); + if (want("agent_run")) parts.push(agentSource(f, n)); + if (want("audit_event")) parts.push(auditSource(f, n)); + if (!parts.length) return { items: [], nextCursor: null }; + + // 表达式很长,估算代价会触发 JIT,编译一次就要上秒;这条查询只取几百行,关掉 JIT + const rows = await db.transaction().execute(async (trx) => { + await sql`set local jit = off`.execute(trx); + return sql`select u.*, (extract(epoch from u.at) * 1000000)::bigint as cursor_us + from (${sql.join( + parts.map((p) => sql`(${p})`), + sql` union all `, + )}) u + order by u.at desc, u.id desc limit ${n}`.execute(trx); + }); const items = rows.rows.slice(0, q.limit); const last = items[items.length - 1]; return { diff --git a/apps/server/test/logs/legacy-query.ts b/apps/server/test/logs/legacy-query.ts new file mode 100644 index 00000000..9404a491 --- /dev/null +++ b/apps/server/test/logs/legacy-query.ts @@ -0,0 +1,232 @@ +import type { LogSource, LogType } from "@umt/contracts"; +import { type DB, type Kysely, sql } from "@umt/db"; +import type { RawBuilder } from "kysely"; +import { BUSINESS_EXPR } from "../../src/modules/auth-settings/engine-business.ts"; +import { + BUSINESS_OBJECT_TYPE, + decodeCursor, + encodeCursor, + type LogQueries, + toEntry, +} from "../../src/modules/logs/query.ts"; + +// 改写前的日志查询(先构造带派生列的子查询、外层再筛选),只作对照口径:新查询在同一份数据上的结果必须与它一致。 +// 在大数据量上极慢,只用于小数据集。 + +type Row = Parameters[0]; + +const SOURCE_TYPES: Record = { + engine_run: ["task"], + schedule_run: ["auto"], + agent_run: ["auto"], + audit_event: ["auth", "audit"], +}; + +const escapeLike = (s: string) => s.replace(/[\\%_]/g, (c) => `\\${c}`); + +const UUID_RE = "^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$"; + +function engineSource(schema: string): RawBuilder { + if (!/^[a-z_][a-z0-9_]*$/.test(schema)) throw new Error(`非法 schema 名 ${schema}`); + const t = sql.raw(`"${schema}".workflow_runs`); + const kind = sql`(biz.b->>'kind')`; + const ref = sql`(biz.b->>'refId')`; + const uuidRef = sql`(case when ${ref} ~* ${UUID_RE} then ${ref}::uuid end)`; + const objectType = sql`(case ${kind} ${sql.join( + Object.entries(BUSINESS_OBJECT_TYPE).map(([k, v]) => sql`when ${sql.lit(k)} then ${sql.lit(v)}`), + sql` `, + )} end)`; + const title = sql`(case ${kind} + when 'schedule' then (select j.name from scheduled_job j where j.id = ${ref}) + when 'agent' then (select a.goal from agent_run a where a.id = ${uuidRef}) + when 'download' then (select d.name from download_task d where d.id = ${uuidRef}) + when 'organization' then (select coalesce(e.title, r.target_path, r.source_path) from organization_record r + left join media_entity e on e.id = r.media_entity_id where r.id = ${uuidRef}) + when 'subscription' then (select coalesce(e.title, s.title_query) from subscription s + left join media_entity e on e.id = s.media_entity_id where s.id = ${uuidRef}) + when 'feed' then (select f.name from feed_source f where f.id = ${uuidRef}) + when 'site' then (select s.name from site s where s.id = ${uuidRef}) + when 'decision' then (select d.title from decision d where d.id = ${uuidRef}) + when 'media' then (select e.title from media_entity e where e.id = ${uuidRef}) + end)`; + const agent = sql`(case when ${kind} = 'agent' then + (select jsonb_build_object('status', a.status, 'trigger', a.trigger) from agent_run a where a.id = ${uuidRef}) end)`; + const st = sql.raw("workflow_runs.status"); + return sql`select + 'engine_run:' || workflow_runs.id as id, + 'task'::text as type, + 'engine_run'::text as source, + workflow_runs.id::text as ref_id, + workflow_runs.created_at as at, + workflow_runs.finished_at as finished_at, + (case when workflow_runs.finished_at is not null and workflow_runs.started_at is not null + then extract(epoch from workflow_runs.finished_at - workflow_runs.started_at) * 1000 end)::float8 as duration_ms, + (case + when ${st} = 'failed' then 'error' + when ${st} = 'canceled' then 'warn' + when ${st} in ('completed', 'succeeded') then (case ag.a->>'status' + when 'failed' then 'error' when 'budget_exceeded' then 'warn' when 'cancelled' then 'warn' else 'success' end) + else 'running' end) as level, + ${st}::text as status, + coalesce(bt.title, workflow_runs.workflow_name) as title, + (case when ${st} = 'failed' then left(coalesce(workflow_runs.error->>'message', workflow_runs.error::text), 500) end) + as message, + ${objectType} as object_type, + ${ref} as object_id, + bt.title as object_label, + null::text as actor_type, + null::text as actor_id, + jsonb_build_object('workflowName', workflow_runs.workflow_name, 'businessKind', ${kind}, + 'businessRefId', ${ref}, 'businessTitle', bt.title, 'outcome', ag.a->>'status', + 'agentTrigger', ag.a->>'trigger', 'attempts', workflow_runs.attempts) as extra, + (biz.b is null + or coalesce(${kind} = 'schedule' and ${st} in ('completed', 'succeeded'), false) + or coalesce((ag.a->>'trigger') in ('event', 'schedule'), false)) as routine, + concat_ws(' ', workflow_runs.workflow_name, workflow_runs.id, bt.title, ${ref}, + left(workflow_runs.error::text, 500)) as search + from ${t} + cross join lateral (select ${BUSINESS_EXPR} as b) biz + cross join lateral (select ${title} as title) bt + cross join lateral (select ${agent} as a) ag + where workflow_runs.namespace_id = 'default'`; +} + +const SCHEDULE_SOURCE = sql`select + 'schedule_run:' || r.id::text as id, + 'auto'::text as type, + 'schedule_run'::text as source, + r.id::text as ref_id, + r.started_at as at, + r.finished_at as finished_at, + r.duration_ms::float8 as duration_ms, + (case + when r.status = 'failed' or r.agent_status = 'failed' then 'error' + when r.agent_status in ('budget_exceeded', 'cancelled') then 'warn' + when r.status = 'ok' then 'success' + when r.status = 'skipped' then 'info' + else 'running' end) as level, + r.status::text as status, + coalesce(j.name, r.job_id) as title, + coalesce(r.error, r.summary) as message, + 'scheduled_job'::text as object_type, + r.job_id as object_id, + j.name as object_label, + r.trigger::text as actor_type, + null::text as actor_id, + jsonb_build_object('jobId', r.job_id, 'jobKind', j.kind, 'trigger', r.trigger, 'agentRunId', r.agent_run_id, + 'agentStatus', r.agent_status, 'summary', r.summary, 'error', r.error) as extra, + (r.status in ('ok', 'skipped') and coalesce(j.kind::text, 'task') <> 'agent_check' + and r.agent_status is distinct from 'failed') as routine, + concat_ws(' ', r.job_id, j.name, r.summary, r.error, r.agent_run_id::text) as search + from scheduled_job_run r left join scheduled_job j on j.id = r.job_id`; + +const AGENT_SOURCE = sql`select + 'agent_run:' || a.id::text as id, + 'auto'::text as type, + 'agent_run'::text as source, + a.id::text as ref_id, + a.created_at as at, + a.finished_at as finished_at, + (case when a.finished_at is not null and a.started_at is not null + then extract(epoch from a.finished_at - a.started_at) * 1000 end)::float8 as duration_ms, + (case a.status + when 'succeeded' then 'success' when 'failed' then 'error' + when 'cancelled' then 'warn' when 'budget_exceeded' then 'warn' else 'running' end) as level, + a.status::text as status, + a.goal as title, + coalesce(a.summary, a.error->>'message', a.spawn_reason) as message, + 'agent_run'::text as object_type, + a.id::text as object_id, + null::text as object_label, + 'event'::text as actor_type, + a.created_by as actor_id, + jsonb_build_object('trigger', a.trigger, 'mode', a.mode, 'toolCalls', a.tool_call_count, + 'spawnReason', a.spawn_reason, 'model', a.model) as extra, + false as routine, + concat_ws(' ', a.goal, a.summary, a.id::text, left(a.error::text, 500)) as search + from agent_run a where a.trigger = 'event' and a.parent_run_id is null`; + +const AUDIT_SOURCE = sql`select + 'audit_event:' || e.id::text as id, + (case when e.action like 'admin.%' or e.action like 'api_key.%' then 'auth' else 'audit' end) as type, + 'audit_event'::text as source, + e.id::text as ref_id, + e.at as at, + null::timestamptz as finished_at, + null::float8 as duration_ms, + (case + when e.action in ('admin.login_failed', 'capability.denied') then 'warn' + when e.action in ('admin.login_success', 'admin.passkey_login', 'admin.setup') then 'success' + else 'info' end) as level, + e.action as status, + e.action as title, + (case when e.action like 'admin.%' then concat_ws(' · ', e.payload->>'username', e.payload->>'ip') + else nullif(concat_ws(' · ', e.payload->>'reason', e.payload->>'error'), '') end) as message, + e.target_type as object_type, + e.target_id as object_id, + null::text as object_label, + e.actor_type as actor_type, + e.actor_id as actor_id, + jsonb_build_object('action', e.action, 'traceId', e.trace_id, 'payload', e.payload) as extra, + false as routine, + concat_ws(' ', e.action, e.target_type, e.target_id, e.actor_id, left(e.payload::text, 2000)) as search + from audit_event e`; + +export function legacyLogQueries(db: Kysely, opts: { engineSchema?: string } = {}): LogQueries { + const engineSchema = opts.engineSchema ?? "openworkflow"; + let engineReady = false; + const engineExists = async () => { + if (engineReady) return true; + const r = await sql<{ + t: string | null; + }>`select to_regclass(${`"${engineSchema}".workflow_runs`})::text as t`.execute(db); + engineReady = !!r.rows[0]?.t; + return engineReady; + }; + + return { + async list(q) { + const types = q.type?.length ? new Set(q.type) : null; + const sources: [LogSource, RawBuilder][] = []; + const want = (s: LogSource) => !types || SOURCE_TYPES[s].some((t) => types.has(t)); + if (want("engine_run") && (await engineExists())) sources.push(["engine_run", engineSource(engineSchema)]); + if (want("schedule_run")) sources.push(["schedule_run", SCHEDULE_SOURCE]); + if (want("agent_run")) sources.push(["agent_run", AGENT_SOURCE]); + if (want("audit_event")) sources.push(["audit_event", AUDIT_SOURCE]); + if (!sources.length) return { items: [], nextCursor: null }; + + const conds: RawBuilder[] = []; + if (types) conds.push(sql`s.type = any(${[...types]}::text[])`); + if (q.level?.length) conds.push(sql`s.level = any(${q.level}::text[])`); + if (q.from) conds.push(sql`s.at >= ${new Date(q.from)}`); + if (q.to) conds.push(sql`s.at < ${new Date(q.to)}`); + if (q.q) conds.push(sql`s.search ilike ${`%${escapeLike(q.q)}%`}`); + if (q.objectType) conds.push(sql`s.object_type = ${q.objectType}`); + if (q.objectId) conds.push(sql`s.object_id = ${q.objectId}`); + if (q.system !== "1" && !q.objectId) conds.push(sql`not s.routine`); + const cur = decodeCursor(q.cursor); + if (cur) { + conds.push(sql`(s.at, s.id) < (timestamptz 'epoch' + ${cur.us} * interval '1 microsecond', ${cur.id})`); + } + const where = conds.length ? sql`where ${sql.join(conds, sql` and `)}` : sql``; + const n = q.limit + 1; + const parts = sources.map( + ([, src]) => sql`(select s.* from (${src}) s ${where} order by s.at desc, s.id desc limit ${n})`, + ); + // 对照只关心结果;关掉 JIT 省掉每条查询上秒的编译 + const rows = await db.transaction().execute(async (trx) => { + await sql`set local jit = off`.execute(trx); + return sql`select u.*, (extract(epoch from u.at) * 1000000)::bigint as cursor_us + from (${sql.join(parts, sql` union all `)}) u + order by u.at desc, u.id desc limit ${n}`.execute(trx); + }); + const items = rows.rows.slice(0, q.limit); + const last = items[items.length - 1]; + return { + items: items.map(toEntry), + nextCursor: + rows.rows.length > q.limit && last ? encodeCursor({ us: Number(last.cursor_us), id: last.id }) : null, + }; + }, + }; +} diff --git a/apps/server/test/logs/logs-legacy-parity.test.ts b/apps/server/test/logs/logs-legacy-parity.test.ts new file mode 100644 index 00000000..f7d709bb --- /dev/null +++ b/apps/server/test/logs/logs-legacy-parity.test.ts @@ -0,0 +1,117 @@ +import type { LogPage } from "@umt/contracts"; +import { sql } from "@umt/db"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { type LogQueryInput, logQueries } from "../../src/modules/logs/query.ts"; +import { createHarness, type Harness, pgAvailable } from "../auth-settings/harness.ts"; +import { legacyLogQueries } from "./legacy-query.ts"; +import { seedLogVolume } from "./volume-seed.ts"; + +// 改写后的日志查询与改写前(先算派生列再筛选)在同一份小数据上逐条一致:筛选、关键词、对象、时间、游标分页。 + +const hasPg = await pgAvailable(); + +describe.skipIf(!hasPg)("日志查询改写前后口径一致", () => { + let h: Harness; + let feedId = ""; + let downloadId = ""; + let agentId = ""; + + beforeAll(async () => { + h = await createHarness({ engine: true }); + await seedLogVolume(h.db, h.engineSchema, 400); + const wr = sql.raw(`"${h.engineSchema}".workflow_runs`); + feedId = (await sql<{ id: string }>`select id::text from feed_source where name like 'my%'`.execute(h.db)).rows[0]! + .id; + downloadId = (await sql<{ id: string }>`select id::text from download_task order by name limit 1`.execute(h.db)) + .rows[0]!.id; + agentId = ( + await sql<{ id: string }>`select id::text from agent_run where status = 'failed' order by goal limit 1`.execute( + h.db, + ) + ).rows[0]!.id; + // 没有业务 id 的运行靠业务任务行(task.engine_run_id)关联对象:一条指向失败的智能体会话、一条指向下载任务 + await sql`insert into task (kind, idempotency_key, engine_run_id, subject_type, subject_id) + values ('agent', 'parity-1', 'vol-000060', 'agent_run', ${agentId}), + ('download', 'parity-2', 'vol-000061', 'download_task', ${downloadId})`.execute(h.db); + // 失败的智能体会话对应的引擎运行(完成)按智能体结果算失败;定时任务的手动重跑不算定时触发 + await sql`update ${wr} set input = jsonb_build_object('input', jsonb_build_object('agentRunId', ${agentId}::text)) + where id = 'vol-000077'`.execute(h.db); + await sql`insert into tasks_run_meta (run_id, task_name, task_version, idempotency_key) + values ('vol-000089', 'purgeTrash', 1, 'manual:1')`.execute(h.db); + // 审计里指向下载任务的一条(按对象筛选时与任务一起列出) + await sql`insert into audit_event (actor_type, actor_id, action, target_type, target_id, payload) + values ('agent', 'run-1', 'capability.denied', 'download_task', ${downloadId}, '{"reason":"超出授权"}'::jsonb)`.execute( + h.db, + ); + }, 120_000); + afterAll(async () => { + await h?.close(); + }); + + const at = (hoursAgo: number) => new Date(Date.now() - hoursAgo * 3600_000).toISOString(); + const filters = (): [string, Omit][] => [ + ["全部", {}], + ["含例行", { system: "1" }], + ["error", { level: ["error"] }], + ["warn", { level: ["warn"] }], + ["success", { level: ["success"] }], + ["running + info", { level: ["running", "info"] }], + ["error 含例行", { level: ["error"], system: "1" }], + ["task", { type: ["task"] }], + ["task 含例行", { type: ["task"], system: "1" }], + ["auto", { type: ["auto"], system: "1" }], + ["auth", { type: ["auth"] }], + ["audit", { type: ["audit"] }], + ["task + auth + warn", { type: ["task", "auth"], level: ["warn"] }], + ["近 6 小时", { from: at(6), system: "1" }], + ["6 到 2 小时前", { from: at(6), to: at(2) }], + ["q 业务标题", { q: "feeds", system: "1" }], + ["q 下载名", { q: "下载 1" }], + ["q 智能体目标", { q: "调查 1", system: "1" }], + ["q 运行 id", { q: "VOL-00012", system: "1" }], + ["q 工作流名", { q: "ingest" }], + ["q 错误", { q: "失败原因 9", system: "1" }], + ["q 审计负载", { q: "10.1.0.5" }], + ["q 定时任务", { q: "超时", system: "1" }], + ["q 例行定时", { q: "清理回收站", system: "1" }], + ["q 字面 %", { q: "%" }], + ["q 不命中", { q: "zz-nomatch", system: "1" }], + ["对象 feed", { objectType: "feed_source", objectId: feedId }], + ["对象 下载(只给 id)", { objectId: downloadId }], + ["对象 下载 + 类型", { objectType: "download_task", objectId: downloadId }], + ["对象 定时任务", { objectType: "scheduled_job", objectId: "trash.purge" }], + ["对象 智能体", { objectType: "agent_run", objectId: agentId }], + ["对象类型不存在", { objectType: "nope" }], + ]; + + it("各种筛选一次取全:逐条一致", async () => { + const next = logQueries(h.db, { engineSchema: h.engineSchema }); + const old = legacyLogQueries(h.db, { engineSchema: h.engineSchema }); + const empty: string[] = []; + for (const [name, f] of filters()) { + const a: LogPage = await next.list({ ...f, limit: 200 }); + const b: LogPage = await old.list({ ...f, limit: 200 }); + expect(a, name).toEqual(b); + if (!a.items.length) empty.push(name); + } + // 对照有意义:除了故意不命中的,筛选都有结果 + expect(empty).toEqual(["q 字面 %", "q 不命中", "对象类型不存在"]); + }, 120_000); + + it("游标分页:每一页与改写前一致", async () => { + const next = logQueries(h.db, { engineSchema: h.engineSchema }); + const old = legacyLogQueries(h.db, { engineSchema: h.engineSchema }); + for (const f of [{ system: "1" as const }, { level: ["error" as const] }, { q: "下载", system: "1" as const }]) { + let cursor: string | null = null; + let pages = 0; + do { + const a: LogPage = await next.list({ ...f, limit: 37, cursor }); + const b: LogPage = await old.list({ ...f, limit: 37, cursor }); + expect(a).toEqual(b); + cursor = a.nextCursor; + pages++; + } while (cursor && pages < 40); + expect(pages).toBeGreaterThan(1); + } + }, 120_000); +}); diff --git a/apps/server/test/logs/logs-volume.test.ts b/apps/server/test/logs/logs-volume.test.ts new file mode 100644 index 00000000..c97e6bbc --- /dev/null +++ b/apps/server/test/logs/logs-volume.test.ts @@ -0,0 +1,105 @@ +import { LogPage } from "@umt/contracts"; +import { sql } from "@umt/db"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { createLogStreamPoller, createLogsModule } from "../../src/modules/logs/index.ts"; +import { logQueries } from "../../src/modules/logs/query.ts"; +import { bearer, createHarness, type Harness, pgAvailable } from "../auth-settings/harness.ts"; +import { DEV_KEY, json } from "../helpers.ts"; +import { seedLogVolume, VOLUME_RUNS } from "./volume-seed.ts"; + +// 日志接口在真实数据量上的耗时(约 30,000 条引擎运行 + 定时任务运行 + 审计)。 +// 改写前这几种查询在部署库上超时(> 60s);每个来源先按原始列筛选取前 n 条再算标题后,应在 1 秒内(阈值放宽到 1.5s 防抖动)。 + +const hasPg = await pgAvailable(); +const LIMIT_MS = 1500; + +describe.skipIf(!hasPg)("日志查询耗时(约 3 万条引擎运行)", () => { + let h: Harness; + let admin: ReturnType; + + beforeAll(async () => { + h = await createHarness({ engine: true }); + const { createApp } = await import("../../src/http/app.ts"); + const { createAuthSettingsModule } = await import("../../src/modules/auth-settings/index.ts"); + h.app = createApp({ + config: h.ctx.config, + logger: h.ctx.logger, + bus: h.bus, + db: h.db, + role: "all", + version: "0.0.0-test", + modules: [ + createAuthSettingsModule({ setupTokenOnStart: false, engineSchema: h.engineSchema }), + createLogsModule({ engineSchema: h.engineSchema }), + ], + ctx: h.ctx, + }) as Harness["app"]; + admin = bearer(h.app, DEV_KEY); + await seedLogVolume(h.db, h.engineSchema); + const n = await sql<{ + n: number; + }>`select count(*)::int as n from ${sql.raw(`"${h.engineSchema}".workflow_runs`)}`.execute(h.db); + expect(n.rows[0]!.n).toBe(VOLUME_RUNS); + // 预热连接 + expect((await admin("GET", "/logs?limit=1")).status).toBe(200); + }, 120_000); + afterAll(async () => { + await h?.close(); + }); + + const since24h = () => encodeURIComponent(new Date(Date.now() - 24 * 3600_000).toISOString()); + const cases: [string, () => string, (items: LogPage["items"]) => void][] = [ + [ + "level=error&since=24h", + () => `?level=error&from=${since24h()}&limit=50`, + (items) => { + expect(items.length).toBeGreaterThan(0); + expect(items.every((e) => e.level === "error")).toBe(true); + }, + ], + [ + "system=1&q=feeds", + () => "?system=1&q=feeds&limit=50", + (items) => { + // 业务标题(RSS 源「my feeds 汇总」)命中的任务 + expect(items.some((e) => e.source === "engine_run" && e.title === "my feeds 汇总")).toBe(true); + }, + ], + [ + "type=task&level=error", + () => "?type=task&level=error&limit=50", + (items) => { + expect(items).toHaveLength(50); + expect(items.every((e) => e.type === "task" && e.level === "error")).toBe(true); + }, + ], + ["type=task", () => "?type=task&limit=50", (items) => expect(items).toHaveLength(50)], + ["默认", () => "?limit=50", (items) => expect(items).toHaveLength(50)], + // 关键词一条都不命中:引擎来源要扫完全部运行,是最坏情况 + ["system=1&q=<不命中>", () => "?system=1&q=zz-nomatch&limit=50", (items) => expect(items).toEqual([])], + ]; + + for (const [name, query, check] of cases) { + it(`${name} < ${LIMIT_MS}ms`, async () => { + const t = performance.now(); + const res = await admin("GET", `/logs${query()}`); + const ms = performance.now() - t; + expect(res.status).toBe(200); + const items = LogPage.parse((await json(res)).data).items; + console.log(`[logs-volume] ${name}: ${ms.toFixed(0)}ms(${items.length} 条)`); + check(items); + expect(ms).toBeLessThan(LIMIT_MS); + }); + } + + it(`流的一轮轮询 < ${LIMIT_MS}ms`, async () => { + const poller = createLogStreamPoller(logQueries(h.db, { engineSchema: h.engineSchema }), { system: "1" }); + for (let i = 0; i < 2; i++) { + const t = performance.now(); + await poller.poll(); + const ms = performance.now() - t; + console.log(`[logs-volume] 流轮询 #${i + 1}: ${ms.toFixed(0)}ms`); + expect(ms).toBeLessThan(LIMIT_MS); + } + }); +}); diff --git a/apps/server/test/logs/volume-seed.ts b/apps/server/test/logs/volume-seed.ts new file mode 100644 index 00000000..bddb63b9 --- /dev/null +++ b/apps/server/test/logs/volume-seed.ts @@ -0,0 +1,87 @@ +import { type DB, type Kysely, sql } from "@umt/db"; + +// 日志性能测试的数据量:约 30,000 条引擎运行(多种工作流 / 状态 / 业务对象,约 28 天) +// + 定时任务运行 + 智能体运行 + 审计。分布参照部署库(大部分是 RSS / 站点快照的成功运行,失败约 1%)。 + +export const VOLUME_RUNS = 30_000; + +export async function seedLogVolume(db: Kysely, engineSchema: string, runs = VOLUME_RUNS): Promise { + const wr = sql.raw(`"${engineSchema}".workflow_runs`); + await sql`insert into site (key, name, adapter, base_url) + select 'vs' || g, '站点 ' || g, 'nexusphp', 'https://s' || g || '.example' from generate_series(1, 20) g`.execute( + db, + ); + await sql`insert into feed_source (name, url_display) + select case when g = 7 then 'my feeds 汇总' else 'RSS 源 ' || g end, 'https://f' || g || '.example' + from generate_series(1, 30) g`.execute(db); + await sql`insert into media_entity (kind, title) select 'movie', '电影 ' || g from generate_series(1, 500) g`.execute( + db, + ); + await sql`insert into subscription (media_kind, media_entity_id, title_query) + select 'movie', e.id, null from (select id from media_entity order by title limit 200) e`.execute(db); + const inst = await sql<{ id: string }>`insert into downloader_instance (name, base_url) + values ('qb-volume', 'http://qb') returning id`.execute(db); + await sql`insert into download_task (instance_id, torrent_hash, name) + select ${inst.rows[0]!.id}::uuid, md5('vol' || g) || left(md5('h' || g), 8), '下载 ' || g || ' 2160p' + from generate_series(1, 2000) g`.execute(db); + await sql`insert into agent_run (trigger, goal, status, created_at, started_at, finished_at) + select (array['event', 'schedule', 'user'])[1 + g % 3], '调查 ' || g, + (array['succeeded', 'succeeded', 'succeeded', 'failed', 'budget_exceeded', 'cancelled'])[1 + g % 6], + now() - g * interval '20 minutes', now() - g * interval '20 minutes', now() - g * interval '20 minutes' + interval '1 minute' + from generate_series(1, 1500) g`.execute(db); + await sql`insert into scheduled_job (id, name, board_group, kind, enabled, default_enabled, schedule, default_schedule) + values ('vol.feeds', 'RSS 抓取(量)', 'collect', 'task', true, true, '{"type":"interval","everyMs":60000}'::jsonb, + '{"type":"interval","everyMs":60000}'::jsonb), + ('trash.purge', '清理回收站', 'maintain', 'task', true, true, '{"type":"interval","everyMs":3600000}'::jsonb, + '{"type":"interval","everyMs":3600000}'::jsonb), + ('vol.check', '巡检(量)', 'agent', 'agent_check', true, true, '{"type":"interval","everyMs":86400000}'::jsonb, + '{"type":"interval","everyMs":86400000}'::jsonb) + on conflict (id) do nothing`.execute(db); + await sql`insert into scheduled_job_run (job_id, trigger, status, error, started_at, finished_at, duration_ms) + select (array['vol.feeds', 'trash.purge', 'vol.check'])[1 + g % 3], 'schedule', + (case when g % 50 = 0 then 'failed' when g % 7 = 0 then 'skipped' else 'ok' end), + (case when g % 50 = 0 then '站点超时 ' || g end), + now() - g * interval '37 minutes', now() - g * interval '37 minutes', 500 + from generate_series(1, 1100) g`.execute(db); + await sql`insert into audit_event (at, actor_type, actor_id, action, target_type, target_id, payload) + select now() - g * interval '53 minutes', 'user', 'admin', + (array['admin.login_success', 'settings.update', 'admin.login_failed', 'download.cancel'])[1 + g % 4], + 'settings', 'k' || g, jsonb_build_object('ip', '10.1.0.' || (g % 200)) + from generate_series(1, 800) g`.execute(db); + + // 引擎运行:按 g % 100 分桶决定工作流与业务对象,g 越大越早(每条间隔 80 秒) + await sql`with + feeds as (select array_agg(id::text order by name) a from feed_source), + sites as (select array_agg(id::text order by key) a from site where key like 'vs%'), + dls as (select array_agg(id::text order by name) a from download_task where instance_id = ${inst.rows[0]!.id}::uuid), + agents as (select array_agg(id::text order by goal) a from agent_run where goal like '调查 %'), + subs as (select array_agg(id::text order by id) a from subscription), + src as ( + select g, g % 100 as b, now() - g * interval '80 seconds' as at, + (case when g <= 5 then 'running' when g % 97 = 0 then 'failed' when g % 193 = 0 then 'canceled' + else 'completed' end) as st + from generate_series(1, ${runs}) g) + insert into ${wr} (namespace_id, id, workflow_name, version, status, config, context, input, error, attempts, + available_at, started_at, finished_at, created_at, updated_at) + select 'default', 'vol-' || lpad(g::text, 6, '0'), + (case when b < 40 then 'ingestFeed' when b < 55 then 'snapshotSite' when b < 65 then 'refreshSiteAccount' + when b < 70 then 'submitDownload' when b < 75 then 'verifyAndImport' when b < 80 then 'runAgent' + when b < 88 then 'searchSubscription' when b < 93 then 'purgeTrash' when b < 98 then 'scanOrganizeSource' + else 'evaluateAcquisition' end), + 'v1', st, '{}'::jsonb, null, + jsonb_build_object('input', (case + when b < 40 then jsonb_build_object('feedSourceId', feeds.a[1 + g % 30]) + when b < 55 then jsonb_build_object('siteId', sites.a[1 + g % 20]) + when b < 65 then '{}'::jsonb + when b < 75 then jsonb_build_object('downloadTaskId', dls.a[1 + g % 2000]) + when b < 80 then jsonb_build_object('agentRunId', agents.a[1 + g % 1500]) + when b < 88 then jsonb_build_object('subscriptionId', subs.a[1 + g % 200]) + when b < 98 then '{}'::jsonb + else jsonb_build_object('acquisitionPlanId', md5('plan' || g)::uuid::text) end)), + (case when st = 'failed' then jsonb_build_object('message', '失败原因 ' || g) end), + 1, at, at, (case when st = 'running' then null else at + interval '3 seconds' end), at, at + from src, feeds, sites, dls, agents, subs`.execute(db); + await sql`analyze ${wr}`.execute(db); + await sql`analyze site, feed_source, media_entity, subscription, download_task, agent_run, scheduled_job, + scheduled_job_run, audit_event`.execute(db); +}