From 75f5d26964cc12f872c0bb14164b20b7db23cc6c Mon Sep 17 00:00:00 2001 From: admin <362324317@qq.com> Date: Wed, 27 May 2026 10:08:59 +0800 Subject: [PATCH] feat: add CloudSearch observability --- source_clean/src/config/index.ts | 2 + source_clean/src/database/database.ts | 3 + source_clean/src/main.ts | 2 + .../database-instrumentation.test.ts | 52 +++++++++ .../observability/database-instrumentation.ts | 62 ++++++++++ source_clean/src/observability/psi.test.ts | 52 +++++++++ source_clean/src/observability/psi.ts | 107 ++++++++++++++++++ .../src/observability/slow-query.test.ts | 54 +++++++++ source_clean/src/observability/slow-query.ts | 35 ++++++ 9 files changed, 369 insertions(+) create mode 100644 source_clean/src/observability/database-instrumentation.test.ts create mode 100644 source_clean/src/observability/database-instrumentation.ts create mode 100644 source_clean/src/observability/psi.test.ts create mode 100644 source_clean/src/observability/psi.ts create mode 100644 source_clean/src/observability/slow-query.test.ts create mode 100644 source_clean/src/observability/slow-query.ts diff --git a/source_clean/src/config/index.ts b/source_clean/src/config/index.ts index f114f81..e2a6b1a 100755 --- a/source_clean/src/config/index.ts +++ b/source_clean/src/config/index.ts @@ -25,6 +25,7 @@ export interface Config { uploadDir: string; chromiumPath: string; dbPath: string; + slowQueryThresholdMs: number; } const DEFAULT_JWT_SECRETS = ['CHANGEME-jwt-placeholder-1', 'CHANGEME-jwt-placeholder-2']; @@ -94,6 +95,7 @@ const config: Config = { uploadDir: process.env.UPLOAD_DIR || '/app/uploads', chromiumPath: process.env.CHROMIUM_PATH || "/usr/bin/chromium-browser", dbPath: process.env.DB_PATH || './data/cloudsearch.db', + slowQueryThresholdMs: parseInt(process.env.SLOW_QUERY_THRESHOLD_MS || '100', 10), }; // Startup validation done by startup-validator diff --git a/source_clean/src/database/database.ts b/source_clean/src/database/database.ts index c5b1c1c..e360bc9 100755 --- a/source_clean/src/database/database.ts +++ b/source_clean/src/database/database.ts @@ -3,6 +3,8 @@ import path from 'path'; import bcrypt from 'bcryptjs'; import config from '../config'; import { formatLocalDateTime } from '../utils/time'; +import { createSlowQueryObserver } from '../observability/slow-query'; +import { instrumentDatabaseSlowQueries } from '../observability/database-instrumentation'; let db: Database.Database | null = null; @@ -16,6 +18,7 @@ export function getDb(): Database.Database { } db = new Database(config.dbPath); + instrumentDatabaseSlowQueries(db, createSlowQueryObserver({ thresholdMs: config.slowQueryThresholdMs })); db.pragma('journal_mode = WAL'); db.pragma('foreign_keys = ON'); diff --git a/source_clean/src/main.ts b/source_clean/src/main.ts index 9ab05a5..40e353f 100755 --- a/source_clean/src/main.ts +++ b/source_clean/src/main.ts @@ -15,6 +15,7 @@ import userRoutes from './user/routes'; import { pansouWebProxy } from './proxy/pansou-web'; import { checkAndRunScheduledCleanup } from './cloud/cleanup.service'; import { refreshAllStorageInfo } from './cloud/cloud.service'; +import { readPsiSummary } from './observability/psi'; const app = express(); @@ -127,6 +128,7 @@ app.get('/health', async (_req, res) => { pansou: pansouStatus, videoParser: videoParserStatus, }, + pressure: readPsiSummary(), }); }); diff --git a/source_clean/src/observability/database-instrumentation.test.ts b/source_clean/src/observability/database-instrumentation.test.ts new file mode 100644 index 0000000..f7a6a01 --- /dev/null +++ b/source_clean/src/observability/database-instrumentation.test.ts @@ -0,0 +1,52 @@ +import { describe, expect, it, vi } from "vitest"; +import { instrumentDatabaseSlowQueries } from "./database-instrumentation"; + +describe("database slow-query instrumentation", () => { + it("wraps prepared statement methods and records their duration", () => { + const record = vi.fn(); + const statement = { + get: vi.fn(() => ({ ok: true })), + all: vi.fn(() => [1, 2]), + run: vi.fn(() => ({ changes: 1 })), + }; + const db = { + prepare: vi.fn(() => statement), + }; + + instrumentDatabaseSlowQueries(db as any, { record, now: (() => { + const ticks = [100, 155, 200, 204]; + return () => ticks.shift() ?? 204; + })() }); + + const wrapped = (db.prepare as any)("SELECT * FROM search_stats WHERE keyword = ?"); + expect(wrapped.get("x")).toEqual({ ok: true }); + expect(wrapped.all()).toEqual([1, 2]); + + expect(record).toHaveBeenCalledWith({ + sql: "SELECT * FROM search_stats WHERE keyword = ?", + method: "get", + durationMs: 55, + }); + expect(record).toHaveBeenCalledWith({ + sql: "SELECT * FROM search_stats WHERE keyword = ?", + method: "all", + durationMs: 4, + }); + }); + + it("is idempotent", () => { + const record = vi.fn(); + const statement = { get: vi.fn(() => 1) }; + const originalPrepare = vi.fn(() => statement); + const db = { prepare: originalPrepare }; + + instrumentDatabaseSlowQueries(db as any, { record, now: () => 1 }); + const wrappedPrepare = db.prepare; + instrumentDatabaseSlowQueries(db as any, { record, now: () => 1 }); + + (db.prepare as any)("SELECT 1").get(); + + expect(db.prepare).toBe(wrappedPrepare); + expect(originalPrepare).toHaveBeenCalledTimes(1); + }); +}); diff --git a/source_clean/src/observability/database-instrumentation.ts b/source_clean/src/observability/database-instrumentation.ts new file mode 100644 index 0000000..2ca49ff --- /dev/null +++ b/source_clean/src/observability/database-instrumentation.ts @@ -0,0 +1,62 @@ +export interface StatementLike { + get?: (...args: any[]) => any; + all?: (...args: any[]) => any; + run?: (...args: any[]) => any; + iterate?: (...args: any[]) => any; + [key: string]: any; +} + +export interface DatabaseLike { + prepare: (sql: string, ...args: any[]) => StatementLike; + [key: string]: any; +} + +export interface QueryRecorder { + record(record: { sql: string; method: string; durationMs: number }): void; +} + +export interface InstrumentationOptions { + record: QueryRecorder["record"]; + now?: () => number; +} + +const INSTRUMENTED = Symbol.for("cloudsearch.db.slowQueryInstrumented"); + +function wrapStatementMethod( + statement: StatementLike, + sql: string, + method: "get" | "all" | "run" | "iterate", + options: Required, +): void { + const original = statement[method]; + if (typeof original !== "function") return; + + statement[method] = function wrappedStatementMethod(this: StatementLike, ...args: any[]) { + const start = options.now(); + try { + return original.apply(this, args); + } finally { + options.record({ sql, method, durationMs: options.now() - start }); + } + }; +} + +export function instrumentDatabaseSlowQueries(db: DatabaseLike, options: InstrumentationOptions): void { + if ((db as any)[INSTRUMENTED]) return; + (db as any)[INSTRUMENTED] = true; + + const completeOptions: Required = { + record: options.record, + now: options.now ?? (() => Number(process.hrtime.bigint()) / 1_000_000), + }; + + const originalPrepare = db.prepare.bind(db); + db.prepare = ((sql: string, ...args: any[]) => { + const statement = originalPrepare(sql, ...args); + wrapStatementMethod(statement, sql, "get", completeOptions); + wrapStatementMethod(statement, sql, "all", completeOptions); + wrapStatementMethod(statement, sql, "run", completeOptions); + wrapStatementMethod(statement, sql, "iterate", completeOptions); + return statement; + }) as DatabaseLike["prepare"]; +} diff --git a/source_clean/src/observability/psi.test.ts b/source_clean/src/observability/psi.test.ts new file mode 100644 index 0000000..82b92cc --- /dev/null +++ b/source_clean/src/observability/psi.test.ts @@ -0,0 +1,52 @@ +import { describe, expect, it } from "vitest"; +import { parsePressureFile, summarizePsi } from "./psi"; + +describe("PSI parser", () => { + it("parses Linux pressure stall information", () => { + const parsed = parsePressureFile("some avg10=0.12 avg60=0.34 avg300=1.23 total=4567\nfull avg10=0.01 avg60=0.02 avg300=0.03 total=89\n"); + + expect(parsed.some).toEqual({ avg10: 0.12, avg60: 0.34, avg300: 1.23, total: 4567 }); + expect(parsed.full).toEqual({ avg10: 0.01, avg60: 0.02, avg300: 0.03, total: 89 }); + }); + + it("summarizes missing PSI files as unavailable instead of throwing", () => { + const summary = summarizePsi({ + cpu: "some avg10=0.00 avg60=0.00 avg300=0.00 total=1\n", + memory: null, + io: null, + }); + + expect(summary.available).toBe(false); + expect(summary.cpu?.some?.avg10).toBe(0); + expect(summary.memory).toBeUndefined(); + expect(summary.io).toBeUndefined(); + }); + + it("marks pressure as warn when avg10 crosses warning thresholds", () => { + const summary = summarizePsi( + { + cpu: "some avg10=25.00 avg60=1.00 avg300=0.50 total=1\nfull avg10=0.00 avg60=0.00 avg300=0.00 total=0\n", + memory: "some avg10=0.00 avg60=0.00 avg300=0.00 total=0\nfull avg10=0.00 avg60=0.00 avg300=0.00 total=0\n", + io: "some avg10=0.00 avg60=0.00 avg300=0.00 total=0\nfull avg10=0.00 avg60=0.00 avg300=0.00 total=0\n", + }, + { cpuSomeAvg10Warn: 20, memorySomeAvg10Warn: 5, ioSomeAvg10Warn: 10 }, + ); + + expect(summary.status).toBe("warn"); + expect(summary.warnings).toContain("cpu.some.avg10 25 >= 20"); + }); + + it("keeps pressure ok when all avg10 values are below thresholds", () => { + const summary = summarizePsi( + { + cpu: "some avg10=0.10 avg60=0.00 avg300=0.00 total=1\nfull avg10=0.00 avg60=0.00 avg300=0.00 total=0\n", + memory: "some avg10=0.00 avg60=0.00 avg300=0.00 total=0\nfull avg10=0.00 avg60=0.00 avg300=0.00 total=0\n", + io: "some avg10=0.00 avg60=0.00 avg300=0.00 total=0\nfull avg10=0.00 avg60=0.00 avg300=0.00 total=0\n", + }, + { cpuSomeAvg10Warn: 20, memorySomeAvg10Warn: 5, ioSomeAvg10Warn: 10 }, + ); + + expect(summary.status).toBe("ok"); + expect(summary.warnings).toEqual([]); + }); +}); diff --git a/source_clean/src/observability/psi.ts b/source_clean/src/observability/psi.ts new file mode 100644 index 0000000..8a9d924 --- /dev/null +++ b/source_clean/src/observability/psi.ts @@ -0,0 +1,107 @@ +import fs from "fs"; + +export interface PsiLine { + avg10: number; + avg60: number; + avg300: number; + total: number; +} + +export interface PressureStats { + some?: PsiLine; + full?: PsiLine; +} + +export interface PsiThresholds { + cpuSomeAvg10Warn?: number; + memorySomeAvg10Warn?: number; + ioSomeAvg10Warn?: number; +} + +export interface PsiSummary { + available: boolean; + status: "ok" | "warn" | "unavailable"; + warnings: string[]; + cpu?: PressureStats; + memory?: PressureStats; + io?: PressureStats; +} + +const DEFAULT_THRESHOLDS: Required = { + cpuSomeAvg10Warn: Number(process.env.PSI_CPU_SOME_AVG10_WARN || 20), + memorySomeAvg10Warn: Number(process.env.PSI_MEMORY_SOME_AVG10_WARN || 5), + ioSomeAvg10Warn: Number(process.env.PSI_IO_SOME_AVG10_WARN || 10), +}; + +function parsePsiLine(line: string): ["some" | "full", PsiLine] | null { + const parts = line.trim().split(/\s+/); + const type = parts.shift(); + if (type !== "some" && type !== "full") return null; + + const values: Record = {}; + for (const part of parts) { + const [key, raw] = part.split("="); + const value = Number(raw); + if (Number.isFinite(value)) values[key] = value; + } + + return [type, { + avg10: values.avg10 ?? 0, + avg60: values.avg60 ?? 0, + avg300: values.avg300 ?? 0, + total: values.total ?? 0, + }]; +} + +export function parsePressureFile(content: string): PressureStats { + const stats: PressureStats = {}; + for (const line of content.split("\n")) { + if (!line.trim()) continue; + const parsed = parsePsiLine(line); + if (parsed) stats[parsed[0]] = parsed[1]; + } + return stats; +} + +function evaluatePressure(summary: PsiSummary, thresholds: Required): void { + const checks: Array<{ label: string; value?: number; threshold: number }> = [ + { label: "cpu.some.avg10", value: summary.cpu?.some?.avg10, threshold: thresholds.cpuSomeAvg10Warn }, + { label: "memory.some.avg10", value: summary.memory?.some?.avg10, threshold: thresholds.memorySomeAvg10Warn }, + { label: "io.some.avg10", value: summary.io?.some?.avg10, threshold: thresholds.ioSomeAvg10Warn }, + ]; + + for (const check of checks) { + if (check.value !== undefined && check.value >= check.threshold) { + summary.warnings.push(`${check.label} ${check.value} >= ${check.threshold}`); + } + } + summary.status = summary.available ? (summary.warnings.length > 0 ? "warn" : "ok") : "unavailable"; +} + +export function summarizePsi( + files: { cpu?: string | null; memory?: string | null; io?: string | null }, + thresholds: PsiThresholds = {}, +): PsiSummary { + const summary: PsiSummary = { available: true, status: "ok", warnings: [] }; + for (const key of ["cpu", "memory", "io"] as const) { + const content = files[key]; + if (!content) { + summary.available = false; + continue; + } + summary[key] = parsePressureFile(content); + } + evaluatePressure(summary, { ...DEFAULT_THRESHOLDS, ...thresholds }); + return summary; +} + +export function readPsiSummary(basePath = "/proc/pressure"): PsiSummary { + const read = (name: string): string | null => { + try { + return fs.readFileSync(`${basePath}/${name}`, "utf8"); + } catch { + return null; + } + }; + return summarizePsi({ cpu: read("cpu"), memory: read("memory"), io: read("io") }); +} diff --git a/source_clean/src/observability/slow-query.test.ts b/source_clean/src/observability/slow-query.test.ts new file mode 100644 index 0000000..5b93c6c --- /dev/null +++ b/source_clean/src/observability/slow-query.test.ts @@ -0,0 +1,54 @@ +import { describe, expect, it, vi } from "vitest"; +import { createSlowQueryObserver, normalizeSqlForLog } from "./slow-query"; + +describe("slow query observer", () => { + it("normalizes SQL before logging", () => { + expect(normalizeSqlForLog("SELECT *\nFROM search_stats WHERE keyword = ? LIMIT 1")).toBe( + "SELECT * FROM search_stats WHERE keyword = ? LIMIT 1", + ); + }); + + it("does not log queries below threshold", () => { + const warn = vi.fn(); + const observer = createSlowQueryObserver({ thresholdMs: 50, warn }); + + observer.record({ sql: "SELECT 1", method: "get", durationMs: 49.4 }); + + expect(warn).not.toHaveBeenCalled(); + }); + + it("logs queries at or above threshold with method, duration and SQL", () => { + const warn = vi.fn(); + const observer = createSlowQueryObserver({ thresholdMs: 50, warn }); + + observer.record({ sql: "SELECT *\nFROM search_stats WHERE keyword = ?", method: "all", durationMs: 51.2 }); + + expect(warn).toHaveBeenCalledTimes(1); + const entry = JSON.parse(warn.mock.calls[0][0]); + expect(entry).toMatchObject({ + event: "slow_query", + method: "all", + durationMs: 51.2, + thresholdMs: 50, + sql: "SELECT * FROM search_stats WHERE keyword = ?", + }); + }); + + it("emits structured JSON for slow query logs", () => { + const warn = vi.fn(); + const observer = createSlowQueryObserver({ thresholdMs: 50, warn }); + + observer.record({ sql: "SELECT * FROM search_stats WHERE keyword = ?", method: "get", durationMs: 75 }); + + const entry = JSON.parse(warn.mock.calls[0][0]); + expect(entry).toMatchObject({ + event: "slow_query", + method: "get", + durationMs: 75, + thresholdMs: 50, + sql: "SELECT * FROM search_stats WHERE keyword = ?", + }); + expect(entry.timestamp).toEqual(expect.any(String)); + }); + +}); diff --git a/source_clean/src/observability/slow-query.ts b/source_clean/src/observability/slow-query.ts new file mode 100644 index 0000000..7c854dc --- /dev/null +++ b/source_clean/src/observability/slow-query.ts @@ -0,0 +1,35 @@ +export type SlowQueryMethod = "get" | "all" | "run" | "iterate" | string; + +export interface SlowQueryRecord { + sql: string; + method: SlowQueryMethod; + durationMs: number; +} + +export interface SlowQueryObserverOptions { + thresholdMs?: number; + warn?: (message: string) => void; +} + +export function normalizeSqlForLog(sql: string): string { + return sql.replace(/\s+/g, " ").trim(); +} + +export function createSlowQueryObserver(options: SlowQueryObserverOptions = {}) { + const thresholdMs = options.thresholdMs ?? Number(process.env.SLOW_QUERY_THRESHOLD_MS || 100); + const warn = options.warn ?? console.warn; + + return { + record({ sql, method, durationMs }: SlowQueryRecord): void { + if (durationMs < thresholdMs) return; + warn(JSON.stringify({ + event: "slow_query", + timestamp: new Date().toISOString(), + method, + durationMs: Number(durationMs.toFixed(1)), + thresholdMs, + sql: normalizeSqlForLog(sql), + })); + }, + }; +}