diff --git a/apps/api/src/modules/caller-analytics/caller-analytics.service.ts b/apps/api/src/modules/caller-analytics/caller-analytics.service.ts index 7736ca9..b2d087a 100644 --- a/apps/api/src/modules/caller-analytics/caller-analytics.service.ts +++ b/apps/api/src/modules/caller-analytics/caller-analytics.service.ts @@ -90,7 +90,7 @@ export class CallerAnalyticsService { dataThrough: h?.payload.dataThrough ?? null, lagMs: h?.payload.lagMs ?? null, durationPrecision: 'milliseconds', health: h?.payload ?? null, summary: analyticsRates(sum), summaryScope: '时间及业务筛选;样本/比例/异常筛选仅影响号码列表', numbers: filtered.slice(q.skip, q.skip + q.take), total: filtered.length, skip: q.skip, take: q.take, - trends: trends.map(normalize), historyNotice: '仅统计新采集链路的数据;旧固定6秒话单不参与应答统计或校准。' }; + trends: trends.map(r => ({ ...normalize(r), time: Number(r.time) })), historyNotice: '仅统计新采集链路的数据;旧固定6秒话单不参与应答统计或校准。' }; } async detail(raw: Query, user: CurrentUser) { const q = parseAnalyticsQuery(raw); if (!q.caller || !q.customerId) throw new BadRequestException('请选择客户和主叫号码'); diff --git a/apps/worker-cdr/src/analytics-main.ts b/apps/worker-cdr/src/analytics-main.ts index 9d4d4e8..1269070 100644 --- a/apps/worker-cdr/src/analytics-main.ts +++ b/apps/worker-cdr/src/analytics-main.ts @@ -46,7 +46,7 @@ async function collect() { } // Persist only after durable queue writes. Crash before checkpoint replays harmlessly. cursor.offset += last + 1; - await writeFile(`${checkpoint}.tmp`, JSON.stringify(cursor), { mode: 0o600 }); await rename(`${checkpoint}.tmp`, checkpoint); + await writeFile(`${checkpoint}.tmp`, JSON.stringify({ ...cursor, gap }), { mode: 0o600 }); await rename(`${checkpoint}.tmp`, checkpoint); } finally { await file.close(); } } async function entries(entries: Array<[string, string[]]>) { @@ -73,7 +73,7 @@ async function observeActive() { } } async function main() { - try { cursor = JSON.parse(await readFile(checkpoint, 'utf8')); } catch { /* First start reads the retained spool. */ } + try { const saved = JSON.parse(await readFile(checkpoint, 'utf8')); cursor = saved; gap = saved.gap === true; } catch { /* First start reads the retained spool. */ } await redis.connect(); await db.$connect(); try { await redis.xgroup('CREATE', ANALYTICS_STREAM, ANALYTICS_GROUP, '0', 'MKSTREAM'); } catch (e) { if (!(e instanceof Error) || !e.message.includes('BUSYGROUP')) throw e; } let pendingCursor = '0-0'; diff --git a/infra/server-b/caller-analytics/deploy.sh b/infra/server-b/caller-analytics/deploy.sh index e0c1cfd..3cdadec 100644 --- a/infra/server-b/caller-analytics/deploy.sh +++ b/infra/server-b/caller-analytics/deploy.sh @@ -21,13 +21,19 @@ for file in /etc/lisglosips/caller-analytics.env /etc/rsyslog.d/35-caller-analyt if test -f "$file"; then cp -a "$file" "$backup/$(basename "$file")"; fi done changed=0 +was_running=$(systemctl is-active lisglosips@caller-analytics || true) rollback() { rc=$? if test "$changed" = 1; then systemctl stop lisglosips@caller-analytics || true cp -a "$backup/opensips.cfg" /etc/opensips/opensips.cfg + for file in /etc/lisglosips/caller-analytics.env /etc/rsyslog.d/35-caller-analytics.conf; do + if test -f "$backup/$(basename "$file")"; then cp -a "$backup/$(basename "$file")" "$file"; else rm -f "$file"; fi + done ln -sfn "$old" /opt/lisglosips/current + systemctl restart rsyslog || true systemctl restart opensips lisglosips@api || true + if test "$was_running" = active; then systemctl start lisglosips@caller-analytics || true; fi fi echo "FAILED rc=$rc backup=$backup (additive analytics tables retained)" exit "$rc" @@ -44,6 +50,8 @@ if grep -q 'CRA1|' "$source_cfg"; then else /usr/bin/node "$new/scripts/instrument-caller-analytics.mjs" "$source_cfg" "$new/opensips-candidate.cfg" fi +python3 "$new/infra/server-b/caller-analytics/verify-hotpath.py" "$new/opensips-candidate.cfg" +systemctl start lisglosips-redis-hotpath-load.service opensips -C -f "$new/opensips-candidate.cfg" set -a source /etc/lisglosips/secrets/mysql-migrate.env diff --git a/infra/server-b/caller-analytics/rsyslog.conf b/infra/server-b/caller-analytics/rsyslog.conf index 9b722bf..f08a61f 100644 --- a/infra/server-b/caller-analytics/rsyslog.conf +++ b/infra/server-b/caller-analytics/rsyslog.conf @@ -1,5 +1,5 @@ # OpenSIPS emits only analytics observations here. No network destination. -if ($programname == 'opensips' and $msg contains 'CRA1|') then { +if (($programname == 'opensips' or $syslogtag startswith '/usr/sbin/opensips[') and $msg contains 'CRA1|') then { action(type="omfile" file="/var/log/lisglosips/caller-analytics.log" fileOwner="syslog" fileGroup="lisglosips" fileCreateMode="0640" queue.type="LinkedList" queue.filename="caller-analytics" diff --git a/infra/server-b/caller-analytics/verify-hotpath.py b/infra/server-b/caller-analytics/verify-hotpath.py new file mode 100644 index 0000000..a9b76d5 --- /dev/null +++ b/infra/server-b/caller-analytics/verify-hotpath.py @@ -0,0 +1,14 @@ +#!/usr/bin/env python3 +"""Align candidate EVALSHA with the installed Lua bytes actually loaded by the loader.""" +import hashlib +from pathlib import Path +import re +import sys +config = Path(sys.argv[1]) +script = Path('/etc/opensips/lisglosips_hotpath.lua').read_text() +sha = hashlib.sha1(script.encode()).hexdigest() +text = config.read_text() +matches = re.findall(r'EVALSHA ([0-9a-f]{40}) 1 cfg:active_version', text) +assert len(matches) == 1, 'Unexpected hotpath layout' +config.write_text(text.replace('EVALSHA ' + matches[0], 'EVALSHA ' + sha)) +print('Installed hotpath SHA:', sha, 'previous:', matches[0]) diff --git a/tests/api/caller-analytics-real.mjs b/tests/api/caller-analytics-real.mjs index 4f7b9e9..c4709e2 100644 --- a/tests/api/caller-analytics-real.mjs +++ b/tests/api/caller-analytics-real.mjs @@ -6,7 +6,7 @@ import { writeFile } from 'node:fs/promises'; import assert from 'node:assert/strict'; const db = new PrismaClient(); const fixture = 'cra_20260831'; -const user = await db.user.findFirst({ where:{ status:'ENABLED', roles:{some:{roleId:'ROLE_SUPER_ADMIN'}} } }); +const user = await db.user.findFirst({ where:{ status:'ENABLED', userRoles:{some:{roleId:'ROLE_SUPER_ADMIN'}} } }); assert(user, 'An existing test administrator is required'); const token = signAccessToken({sub:user.id,username:user.username,roles:['ROLE_SUPER_ADMIN'],typ:'access'}, {secret:process.env.AUTH_ACCESS_TOKEN_SECRET,issuer:process.env.AUTH_TOKEN_ISSUER||'lisglosips-api',audience:process.env.AUTH_TOKEN_AUDIENCE||'lisglosips-web',ttlSeconds:900}); const get = async (path, authenticated = true) => {