From 52c785f6ec4bc04695e9c762b586a290094ffc31 Mon Sep 17 00:00:00 2001 From: kkfluous Date: Fri, 7 Aug 2026 16:05:20 +0800 Subject: [PATCH] feat: observe BI data source health --- docs/bi-refactor-roadmap.md | 2 +- src/server/auth/middleware.ts | 6 +- src/server/index.ts | 2 + src/server/routes/analytics/index.test.ts | 10 +++ src/server/routes/analytics/index.ts | 3 + src/server/routes/energy/cache.ts | 16 +++-- src/server/routes/energy/index.ts | 12 ++-- src/server/routes/mileage/oneos-api.ts | 15 +++-- src/server/source-telemetry.test.ts | 58 +++++++++++++++++ src/server/source-telemetry.ts | 78 +++++++++++++++++++++++ 10 files changed, 182 insertions(+), 20 deletions(-) create mode 100644 src/server/source-telemetry.test.ts create mode 100644 src/server/source-telemetry.ts diff --git a/docs/bi-refactor-roadmap.md b/docs/bi-refactor-roadmap.md index a0e060d..a778a23 100644 --- a/docs/bi-refactor-roadmap.md +++ b/docs/bi-refactor-roadmap.md @@ -172,7 +172,7 @@ SUM(fee) / NULLIF(SUM(kwh), 0) - 轮换已经进入版本历史的氢能数据库凭据。 - 已提供无敏感值的 `.env.example`；测试、生产值由各环境密钥管理。 - 已拆分进程存活检查与无敏感值的配置就绪检查。 -- 增加数据源连通性、最后成功时间和错误率监控。 +- 已增加基于真实业务请求的数据源最后结果、最后成功时间和滚动失败率监控。 ## 9. 每批验收门槛 diff --git a/src/server/auth/middleware.ts b/src/server/auth/middleware.ts index 081334a..5eccab3 100644 --- a/src/server/auth/middleware.ts +++ b/src/server/auth/middleware.ts @@ -31,7 +31,11 @@ export async function authMiddleware(c: Context, next: Next) { } // 跳过不需要认证的路径 - if (path.startsWith('/api/health') || path.startsWith('/api/auth/')) { + if ( + path === '/api/health' + || path === '/api/health/readiness' + || path.startsWith('/api/auth/') + ) { return next(); } diff --git a/src/server/index.ts b/src/server/index.ts index 746cbc2..ff83cf6 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -40,6 +40,8 @@ app.route('/api/vehicle-heatmap', vehicleHeatmapRouter); app.route('/api/hydrogen-heatmap', hydrogenHeatmapRouter); app.route('/api/analytics', analyticsRouter); +app.all('/api/*', (c) => c.json({ error: 'API route not found' }, 404)); + // Serve static files in production app.use('/*', serveStatic({ root: './dist' })); app.use('/*', serveStatic({ root: './dist', path: 'index.html' })); diff --git a/src/server/routes/analytics/index.test.ts b/src/server/routes/analytics/index.test.ts index 83d9ab9..ff994ac 100644 --- a/src/server/routes/analytics/index.test.ts +++ b/src/server/routes/analytics/index.test.ts @@ -50,3 +50,13 @@ test('rejects unpublished metric domains', async () => { assert.equal(response.status, 400); }); + +test('returns observed data-source outcomes without sensitive details', async () => { + const response = await app.request('/data-sources'); + const payload = await response.json(); + + assert.equal(response.status, 200); + assert.equal(payload.semantics, 'observed-requests'); + assert.ok(Array.isArray(payload.sources)); + assert.equal(JSON.stringify(payload).includes('password'), false); +}); diff --git a/src/server/routes/analytics/index.ts b/src/server/routes/analytics/index.ts index de55008..fad4dc4 100644 --- a/src/server/routes/analytics/index.ts +++ b/src/server/routes/analytics/index.ts @@ -5,6 +5,7 @@ import { listMetricDefinitions, type MetricDomain, } from '../../../shared/analytics/catalog.js'; +import { getDataSourceTelemetry } from '../../source-telemetry.js'; const app = new Hono(); @@ -26,4 +27,6 @@ app.get('/metrics', (c) => { }); }); +app.get('/data-sources', (c) => c.json(getDataSourceTelemetry())); + export default app; diff --git a/src/server/routes/energy/cache.ts b/src/server/routes/energy/cache.ts index 5c0fc95..0eb88f2 100644 --- a/src/server/routes/energy/cache.ts +++ b/src/server/routes/energy/cache.ts @@ -1,3 +1,5 @@ +import { observeDataSource, type DataSourceId } from '../../source-telemetry.js'; + /** * SWR 缓存：始终返回热数据，后台定时刷新。 * @@ -70,24 +72,28 @@ async function runRefresh(key: string) { export interface CachedOpts { force?: boolean; + source?: DataSourceId; } export async function cached(key: string, loader: () => Promise, opts: CachedOpts = {}): Promise { + const effectiveLoader = opts.source + ? () => observeDataSource(opts.source!, loader) + : loader; const now = Date.now(); const hit = cache.get(key) as Entry | undefined; if (hit) { hit.lastAccess = now; - hit.loader = loader; + hit.loader = effectiveLoader; } // 强制刷新：等待 loader 完成 if (opts.force) { const ongoing = inflight.get(key) as Promise | undefined; if (ongoing) return ongoing; - const p = loader() + const p = effectiveLoader() .then(value => { const t = Date.now(); - const next: Entry = { value, freshAt: t, expiresAt: t + TTL_MS, loader, lastAccess: t }; + const next: Entry = { value, freshAt: t, expiresAt: t + TTL_MS, loader: effectiveLoader, lastAccess: t }; cache.set(key, next); scheduleRefresh(key, next); return value; @@ -112,10 +118,10 @@ export async function cached(key: string, loader: () => Promise, opts: Cac const ongoing = inflight.get(key) as Promise | undefined; if (ongoing) return ongoing; - const p = loader() + const p = effectiveLoader() .then(value => { const t = Date.now(); - const entry: Entry = { value, freshAt: t, expiresAt: t + TTL_MS, loader, lastAccess: t }; + const entry: Entry = { value, freshAt: t, expiresAt: t + TTL_MS, loader: effectiveLoader, lastAccess: t }; cache.set(key, entry); scheduleRefresh(key, entry); return value; diff --git a/src/server/routes/energy/index.ts b/src/server/routes/energy/index.ts index 82471fb..46a0925 100644 --- a/src/server/routes/energy/index.ts +++ b/src/server/routes/energy/index.ts @@ -389,7 +389,7 @@ app.get('/hydrogen/overview', async (c) => { })); return { kpi, top5, regions, monthly, customers, stations, availableYears, year }; - }, { force }); + }, { force, source: 'hydrogenDatabase' }); return c.json(data); }); @@ -530,7 +530,7 @@ app.get('/hydrogen/daily', async (c) => { // 按日期降序返回 const result = ascDays.slice().sort((a, b) => b.date.localeCompare(a.date)); return result; - }, { force }); + }, { force, source: 'hydrogenDatabase' }); return c.json(data); }); @@ -615,7 +615,7 @@ app.get('/electric/overview', async (c) => { trend: trendArr, latestRecordAt, }; - }, { force }); + }, { force, source: 'electricDatabase' }); return c.json(data); }); @@ -688,7 +688,7 @@ app.get('/electric/orders', async (c) => { totalFee: Math.round((Number(row.totalFee) || 0) * 100) / 100, })), }; - }); + }, { source: 'electricDatabase' }); return c.json(data); }); @@ -779,7 +779,7 @@ app.get('/electric/monthly', async (c) => { }); return months; - }, { force }); + }, { force, source: 'electricDatabase' }); return c.json(data); }); @@ -862,7 +862,7 @@ app.get('/etc/overview', async (c) => { latestTransactionTime: record.latestTransactionTime ? String(record.latestTransactionTime) : null, }, } as const; - }, { force }); + }, { force, source: 'etcDatabase' }); return c.json(data); }); diff --git a/src/server/routes/mileage/oneos-api.ts b/src/server/routes/mileage/oneos-api.ts index 916eb19..b96c38b 100644 --- a/src/server/routes/mileage/oneos-api.ts +++ b/src/server/routes/mileage/oneos-api.ts @@ -4,6 +4,7 @@ import { normalizeOneOsProtocol, type OneOsProtocol, } from './source-policy.js'; +import { observeDataSource } from '../../source-telemetry.js'; dotenv.config(); @@ -313,7 +314,7 @@ export async function fetchOneOsDailyMileage( } else { let pending = inflight.get(cacheKey); if (!pending) { - pending = requestDate(date, protocolPriority).then(result => { + pending = observeDataSource('oneOsMileage', () => requestDate(date, protocolPriority)).then(result => { const ttl = date === shanghaiDate() ? CURRENT_DAY_TTL_MS : HISTORICAL_TTL_MS; cache.delete(cacheKey); if (ttl > 0) { @@ -372,12 +373,12 @@ export async function fetchOneOsMileageDates( ].join(':'); let pending = rangeInflight.get(key); if (!pending) { - pending = requestRange( - startDate, - endDate, - selectedPlates, - protocolPriority, - ).finally(() => rangeInflight.delete(key)); + pending = observeDataSource('oneOsMileage', () => requestRange( + startDate, + endDate, + selectedPlates, + protocolPriority, + )).finally(() => rangeInflight.delete(key)); rangeInflight.set(key, pending); } const rowsByDate = await pending; diff --git a/src/server/source-telemetry.test.ts b/src/server/source-telemetry.test.ts new file mode 100644 index 0000000..8173d94 --- /dev/null +++ b/src/server/source-telemetry.test.ts @@ -0,0 +1,58 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { + clearDataSourceTelemetry, + getDataSourceTelemetry, + observeDataSource, + recordDataSourceOutcome, +} from './source-telemetry.js'; + +test('reports unobserved sources without treating them as failures', () => { + clearDataSourceTelemetry(); + const report = getDataSourceTelemetry(new Date('2026-08-07T08:00:00Z')); + assert.ok(report.sources.every(source => source.state === 'unobserved')); + assert.ok(report.sources.every(source => source.failureRate === null)); +}); + +test('publishes rolling outcomes and last success/failure timestamps', () => { + clearDataSourceTelemetry(); + recordDataSourceOutcome('oneOsMileage', true, 1_000); + recordDataSourceOutcome('oneOsMileage', false, 2_000); + const source = getDataSourceTelemetry().sources.find(item => item.source === 'oneOsMileage'); + assert.deepEqual(source && { + state: source.state, + attempts: source.attempts, + failures: source.failures, + failureRate: source.failureRate, + lastSuccessAt: source.lastSuccessAt, + lastFailureAt: source.lastFailureAt, + }, { + state: 'failing', + attempts: 2, + failures: 1, + failureRate: 0.5, + lastSuccessAt: '1970-01-01T00:00:01.000Z', + lastFailureAt: '1970-01-01T00:00:02.000Z', + }); +}); + +test('observes final operation outcomes without retaining error details', async () => { + clearDataSourceTelemetry(); + await assert.rejects(observeDataSource('hydrogenDatabase', async () => { + throw new Error('sensitive connection detail'); + })); + const serialized = JSON.stringify(getDataSourceTelemetry()); + assert.equal(serialized.includes('sensitive connection detail'), false); +}); + +test('keeps only the latest 100 outcomes', () => { + clearDataSourceTelemetry(); + recordDataSourceOutcome('etcDatabase', false, 1); + for (let index = 0; index < 100; index += 1) { + recordDataSourceOutcome('etcDatabase', true, index + 2); + } + const source = getDataSourceTelemetry().sources.find(item => item.source === 'etcDatabase'); + assert.equal(source?.attempts, 100); + assert.equal(source?.failures, 0); + assert.equal(source?.failureRate, 0); +}); diff --git a/src/server/source-telemetry.ts b/src/server/source-telemetry.ts new file mode 100644 index 0000000..5172535 --- /dev/null +++ b/src/server/source-telemetry.ts @@ -0,0 +1,78 @@ +export const DATA_SOURCE_IDS = [ + 'oneOsMileage', + 'hydrogenDatabase', + 'electricDatabase', + 'etcDatabase', +] as const; + +export type DataSourceId = typeof DATA_SOURCE_IDS[number]; + +interface Outcome { + success: boolean; + at: number; +} + +const MAX_OUTCOMES = 100; +const outcomes = new Map(); +const startedAt = new Date().toISOString(); + +function lastMatching(entries: Outcome[], success: boolean): Outcome | undefined { + for (let index = entries.length - 1; index >= 0; index -= 1) { + if (entries[index].success === success) return entries[index]; + } + return undefined; +} + +export function recordDataSourceOutcome( + source: DataSourceId, + success: boolean, + at = Date.now(), +): void { + const entries = outcomes.get(source) || []; + entries.push({ success, at }); + if (entries.length > MAX_OUTCOMES) entries.splice(0, entries.length - MAX_OUTCOMES); + outcomes.set(source, entries); +} + +export async function observeDataSource( + source: DataSourceId, + operation: () => Promise, +): Promise { + try { + const result = await operation(); + recordDataSourceOutcome(source, true); + return result; + } catch (error) { + recordDataSourceOutcome(source, false); + throw error; + } +} + +export function getDataSourceTelemetry(now = new Date()) { + return { + semantics: 'observed-requests' as const, + windowSize: MAX_OUTCOMES, + processStartedAt: startedAt, + checkedAt: now.toISOString(), + sources: DATA_SOURCE_IDS.map(source => { + const entries = outcomes.get(source) || []; + const failures = entries.filter(entry => !entry.success).length; + const last = entries.at(-1); + const lastSuccess = lastMatching(entries, true); + const lastFailure = lastMatching(entries, false); + return { + source, + state: !last ? 'unobserved' as const : last.success ? 'available' as const : 'failing' as const, + attempts: entries.length, + failures, + failureRate: entries.length > 0 ? Math.round(failures / entries.length * 10_000) / 10_000 : null, + lastSuccessAt: lastSuccess ? new Date(lastSuccess.at).toISOString() : null, + lastFailureAt: lastFailure ? new Date(lastFailure.at).toISOString() : null, + }; + }), + }; +} + +export function clearDataSourceTelemetry(): void { + outcomes.clear(); +}