优化 done-rate 统计

oms/OPTIMIZATION_TRACKER.md

@@ -0,0 +1,142 @@
+# OMS Optimization Tracker
+
+## 状态说明
+
+- `已修正`: 已完成代码或数据层调整。
+- `待修正`: 已确认问题，后续继续跟进。
+- `暂缓`: 已知问题，但当前优先级较低。
+
+## 已修正
+
+### 1. ClickHouse `events` 表增加按月分区
+
+- 状态: `已修正`
+- 背景: 原表无 `PARTITION BY`，`done-rate` 按天统计时会扫描超大范围数据，随着数据量增长容易超时。
+- 处理:
+  - 已完成 `events -> events_v2 -> rename` 的迁移。
+  - 新表使用 `PARTITION BY toYYYYMM(time)`。
+  - 已补齐迁移过程中缺失的 `202511` 和 `202605` 数据。
+- 结果:
+  - 分区剪枝可用于按天/按月统计。
+  - TTL 清理效率提升。
+  - 为后续统计查询优化打下基础。
+
+### 2. `done-rate` 的 ClickHouse 统计查询合并为 1 条聚合查询
+
+- 状态: `已修正`
+- 文件: `services/cron-jobs/done-rate.ts`
+- 原问题:
+  - 原实现分别对 `color_start`、`color_done`、`color_tip` 发 3 次独立查询。
+  - 相同时间范围被重复扫描，成本高。
+- 处理:
+  - 已改为 1 条聚合查询，使用 `uniqIf` 和 `countIf` 一次返回三类统计结果。
+  - 时间边界已修正为 `[day_start, next_day_start)`，避免漏掉当天末尾数据。
+
+### 3. `done-rate` 增加 ClickHouse 查询耗时与结果规模日志
+
+- 状态: `已修正`
+- 文件: `services/cron-jobs/done-rate.ts`
+- 处理:
+  - 增加 ClickHouse 聚合查询耗时日志。
+  - 增加返回行数、有效作品数、无效行数、开始用户数、完成用户数、道具使用次数统计日志。
+- 目标:
+  - 便于线上确认分区和聚合查询优化后的实际收益。
+
+### 4. ClickHouse 分区迁移脚本落库
+
+- 状态: `已修正`
+- 文件: `scripts/migrate-clickhouse-events-partition.sh`
+- 处理:
+  - 支持 Docker 容器内执行 `clickhouse-client`。
+  - 支持历史分区回填、cutover、校验、reconcile-only 缺口补齐。
+  - 修复了若干脚本问题，包括 shell `eval`、Docker stdin、缺口月份补齐逻辑等。
+- 备注:
+  - 该脚本可作为后续类似数据迁移的基线脚本继续演进。
+
+## 待修正
+
+### 1. `done-rate` 后半段 Mongo 聚合更新可能仍然偏慢
+
+- 状态: `待修正`
+- 文件: `services/cron-jobs/done-rate.ts`
+- 现状:
+  - ClickHouse 查询已经优化，但后半段仍然会聚合 `DoneRateModel` 并逐条更新 `TotalDoneRate`。
+  - 当作品数量继续增长时，这一段可能成为新的耗时热点。
+- 可选优化方向:
+  - 改为 `bulkWrite` 批量更新 `TotalDoneRate`。
+  - 评估是否可以用增量方式替代全量聚合。
+  - 增加阶段性耗时日志，拆分 ClickHouse 时间和 Mongo 更新时间。
+
+### 2. `ingestor-service` 先 `ack` 后 flush，存在小窗口数据丢失风险
+
+- 状态: `待修正`
+- 文件: `services/ingestor-service.ts`
+- 现状:
+  - 消息进入内存 buffer 后即 `ack`。
+  - 如果进程在 flush 到 ClickHouse 或 MongoDB 前崩溃，这批消息会丢失。
+- 当前取舍:
+  - 业务上允许少量埋点丢失，因此暂不优先处理。
+- 后续可选方向:
+  - 缩短 flush 间隔。
+  - 在关键批次上改为 flush 成功后再 `ack`。
+  - 引入失败补偿或轻量级死信策略。
+
+### 3. ClickHouse `raw_json_data` 存储成本较高
+
+- 状态: `待修正`
+- 文件: `src/services/clickhouseService.ts`
+- 现状:
+  - 每条事件完整保存原始 JSON。
+  - 同时系统已有 `log-service` 将原始日志写入文件，存在冗余。
+- 风险:
+  - 长期占用大量磁盘。
+  - 增加 merge、备份、迁移成本。
+- 后续可选方向:
+  - 仅保留结构化字段。
+  - 对 `raw_json_data` 做抽样保留。
+  - 只在日志文件中保留完整原文。
+
+### 4. `event-api-service` 缺少限流与更明确的背压策略
+
+- 状态: `待修正`
+- 文件: `services/event-api-service.ts`
+- 现状:
+  - 接口没有基础 rate limit。
+  - RabbitMQ publish buffer 满时仅返回 `503`，没有更系统的保护策略。
+- 风险:
+  - 恶意流量或异常流量可能放大队列和下游压力。
+- 后续可选方向:
+  - 增加基础限流中间件。
+  - 区分客户端错误与系统拥塞。
+  - 评估本地丢弃、采样或重试策略。
+
+### 5. 配置与凭证仍混在代码和部署配置中
+
+- 状态: `待修正`
+- 文件:
+  - `ecosystem.config.js`
+  - `docker-compose.yml`
+  - `docker-compose.prod.yml`
+  - 部分 service 文件中的默认连接串
+- 现状:
+  - 仓库中仍存在明文凭证和默认连接串。
+  - 运行时配置分散，后续维护和切环境成本高。
+- 后续可选方向:
+  - 统一改为 `.env` 或部署平台环境变量管理。
+  - 清理代码中的真实凭证 fallback。
+  - 为生产和开发环境建立更清晰的配置边界。
+
+## 暂缓
+
+### 1. `ingestor-service` 的严格不丢数改造
+
+- 状态: `暂缓`
+- 原因:
+  - 当前业务接受少量埋点丢失。
+  - 与 `done-rate` 和 ClickHouse 性能问题相比，优先级较低。
+
+## 下一步建议
+
+1. 先在线上执行一次 `done-rate`，观察新增的 ClickHouse 耗时日志。
+2. 如果 ClickHouse 已明显变快，再评估 `TotalDoneRate` 的 Mongo 聚合更新是否成为新瓶颈。
+3. 若后续继续做性能治理，优先处理 `done-rate` 的 Mongo 批量更新。

oms/dist/services/cron-jobs/done-rate.js

@@ -38,82 +38,53 @@ async function run(dateStr) {
     const yesterday = targetDay;
     const yesterdayYYYYMMDD = yesterday.format("YYYYMMDD");
     const yesterdayStart = yesterday.toDate();
-    const yesterdayEnd = yesterday.endOf("day").toDate();
+    const nextDayStart = yesterday.add(1, "day").startOf("day").toDate();
     // 格式化日期字符串，使其符合 ClickHouse 的 toDateTime() 函数要求
     const yesterdayStartString = (0, dayjs_1.default)(yesterdayStart).format("YYYY-MM-DD HH:mm:ss");
-    const yesterdayEndString = (0, dayjs_1.default)(yesterdayEnd).format("YYYY-MM-DD HH:mm:ss");
+    const nextDayStartString = (0, dayjs_1.default)(nextDayStart).format("YYYY-MM-DD HH:mm:ss");
     console.log(`[DoneRate Cron] Processing data for date: ${yesterdayYYYYMMDD}`);
     try {
         // --- 1. 从 ClickHouse 中提取数据 (Start, Done, Tip Counts) ---
-        // 1.1 查询目标日期每个作品的独立开始用户数
-        const startCountsQuery = `
+        const aggregateQuery = `
       SELECT
           res,
-          count(DISTINCT uid) AS unique_starts
+          uniqIf(uid, event = 'color_start') AS unique_starts,
+          uniqIf(uid, event = 'color_done') AS unique_dones,
+          countIf(event = 'color_tip') AS tip_count
       FROM ${CLICKHOUSE_EVENTS_TABLE}
-      WHERE event = 'color_start'
+      WHERE event IN ('color_start', 'color_done', 'color_tip')
         AND time >= toDateTime('${yesterdayStartString}')
-        AND time < toDateTime('${yesterdayEndString}')
+        AND time < toDateTime('${nextDayStartString}')
       GROUP BY res
       HAVING res IS NOT NULL
     `;
-        const startResults = await clients_1.clickhouseService.queryEvents(startCountsQuery);
+        const clickhouseQueryStartTime = Date.now();
+        const aggregateResults = await clients_1.clickhouseService.queryEvents(aggregateQuery);
+        const clickhouseQueryElapsedMs = Date.now() - clickhouseQueryStartTime;
         const artworkStartCounts = new Map();
-        startResults.forEach((row) => {
-            if (row.res && mongoose_1.default.Types.ObjectId.isValid(row.res)) {
-                artworkStartCounts.set(row.res, row.unique_starts);
-            }
-            else {
-                console.warn(`[DoneRate Cron] Invalid artwork ID found in start_counts result: ${row.res}`);
-            }
-        });
-        console.log(`[DoneRate Cron] Retrieved ${startResults.length} unique start counts from ClickHouse.`);
-        // 1.2 查询目标日期每个作品的独立完成用户数
-        const doneCountsQuery = `
-      SELECT
-          res,
-          count(DISTINCT uid) AS unique_dones
-      FROM ${CLICKHOUSE_EVENTS_TABLE}
-      WHERE event = 'color_done'
-        AND time >= toDateTime('${yesterdayStartString}')
-        AND time < toDateTime('${yesterdayEndString}')
-      GROUP BY res
-      HAVING res IS NOT NULL
-    `;
-        const doneResults = await clients_1.clickhouseService.queryEvents(doneCountsQuery);
         const artworkDoneCounts = new Map();
-        doneResults.forEach((row) => {
-            if (row.res && mongoose_1.default.Types.ObjectId.isValid(row.res)) {
-                artworkDoneCounts.set(row.res, row.unique_dones);
-            }
-            else {
-                console.warn(`[DoneRate Cron] Invalid artwork ID found in done_counts result: ${row.res}`);
-            }
-        });
-        console.log(`[DoneRate Cron] Retrieved ${doneResults.length} unique done counts from ClickHouse.`);
-        // 1.3 查询目标日期每个作品的使用道具数
-        const tipCountsQuery = `
-      SELECT
-          res,
-          count() AS tip_count
-      FROM ${CLICKHOUSE_EVENTS_TABLE}
-      WHERE event = 'color_tip'
-        AND time >= toDateTime('${yesterdayStartString}')
-        AND time < toDateTime('${yesterdayEndString}')
-      GROUP BY res
-      HAVING res IS NOT NULL
-    `;
-        const tipResults = await clients_1.clickhouseService.queryEvents(tipCountsQuery);
         const artworkTipCounts = new Map();
-        tipResults.forEach((row) => {
+        let invalidArtworkRowCount = 0;
+        aggregateResults.forEach((row) => {
             if (row.res && mongoose_1.default.Types.ObjectId.isValid(row.res)) {
-                artworkTipCounts.set(row.res, row.tip_count);
+                artworkStartCounts.set(row.res, Number(row.unique_starts));
+                artworkDoneCounts.set(row.res, Number(row.unique_dones));
+                artworkTipCounts.set(row.res, Number(row.tip_count));
             }
             else {
-                console.warn(`[DoneRate Cron] Invalid artwork ID found in tip_counts result: ${row.res}`);
+                invalidArtworkRowCount++;
+                console.warn(`[DoneRate Cron] Invalid artwork ID found in aggregate result: ${row.res}`);
             }
         });
-        console.log(`[DoneRate Cron] Retrieved ${tipResults.length} unique tip counts from ClickHouse.`);
+        let totalStartUsers = 0;
+        let totalDoneUsers = 0;
+        let totalTipEvents = 0;
+        aggregateResults.forEach((row) => {
+            totalStartUsers += Number(row.unique_starts) || 0;
+            totalDoneUsers += Number(row.unique_dones) || 0;
+            totalTipEvents += Number(row.tip_count) || 0;
+        });
+        console.log(`[DoneRate Cron] ClickHouse aggregate query completed in ${clickhouseQueryElapsedMs}ms. Rows: ${aggregateResults.length}, valid artworks: ${artworkStartCounts.size}, invalid rows: ${invalidArtworkRowCount}, unique starts: ${totalStartUsers}, unique dones: ${totalDoneUsers}, tip events: ${totalTipEvents}.`);
         // --- 2. 合并数据并更新 DoneRate 模型 (每日记录) ---
         let updatedRecordsCount = 0;
         let createdRecordsCount = 0;

oms/services/cron-jobs/done-rate.ts

@@ -9,26 +9,12 @@ import DoneRateModel from "../../src/models/doneRateModel"; // 导入 DoneRateMo
 const CLICKHOUSE_EVENTS_TABLE = "events"; // 确保与 ClickHouseService 中的表名一致
 
 /**
- * ClickHouse 查询结果接口：每日每个作品的独立开始用户数
+ * ClickHouse 查询结果接口：每日每个作品的开始、完成和道具使用统计。
  */
-interface ClickHouseStartCountResult {
+interface ClickHouseDoneRateAggregateResult {
   res: string; // 作品 ID
   unique_starts: number; // 独立开始用户数
-}
-
-/**
- * ClickHouse 查询结果接口：每日每个作品的独立完成用户数
- */
-interface ClickHouseDoneCountResult {
-  res: string; // 作品 ID
   unique_dones: number; // 独立完成用户数
-}
-
-/**
- * ClickHouse 查询结果接口：每日每个作品的使用道具数 (对应 tipCount)
- */
-interface ClickHouseTipCountResult {
-  res: string; // 作品 ID
   tip_count: number; // 道具使用次数
 }
 
@@ -62,85 +48,61 @@ async function run(dateStr?: string | dayjs.Dayjs): Promise<string> {
   const yesterday = targetDay;
   const yesterdayYYYYMMDD = yesterday.format("YYYYMMDD");
   const yesterdayStart = yesterday.toDate();
-  const yesterdayEnd = yesterday.endOf("day").toDate();
+  const nextDayStart = yesterday.add(1, "day").startOf("day").toDate();
 
   // 格式化日期字符串，使其符合 ClickHouse 的 toDateTime() 函数要求
   const yesterdayStartString = dayjs(yesterdayStart).format("YYYY-MM-DD HH:mm:ss");
-  const yesterdayEndString = dayjs(yesterdayEnd).format("YYYY-MM-DD HH:mm:ss");
+  const nextDayStartString = dayjs(nextDayStart).format("YYYY-MM-DD HH:mm:ss");
 
   console.log(`[DoneRate Cron] Processing data for date: ${yesterdayYYYYMMDD}`);
 
   try {
     // --- 1. 从 ClickHouse 中提取数据 (Start, Done, Tip Counts) ---
-
-    // 1.1 查询目标日期每个作品的独立开始用户数
-    const startCountsQuery = `
+    const aggregateQuery = `
       SELECT
           res,
-          count(DISTINCT uid) AS unique_starts
+          uniqIf(uid, event = 'color_start') AS unique_starts,
+          uniqIf(uid, event = 'color_done') AS unique_dones,
+          countIf(event = 'color_tip') AS tip_count
       FROM ${CLICKHOUSE_EVENTS_TABLE}
-      WHERE event = 'color_start'
+      WHERE event IN ('color_start', 'color_done', 'color_tip')
         AND time >= toDateTime('${yesterdayStartString}')
-        AND time < toDateTime('${yesterdayEndString}')
+        AND time < toDateTime('${nextDayStartString}')
       GROUP BY res
       HAVING res IS NOT NULL
     `;
-    const startResults = await clickhouseService.queryEvents<ClickHouseStartCountResult>(startCountsQuery);
-    const artworkStartCounts = new Map<string, number>();
-    startResults.forEach((row) => {
-      if (row.res && mongoose.Types.ObjectId.isValid(row.res)) {
-        artworkStartCounts.set(row.res, row.unique_starts);
-      } else {
-        console.warn(`[DoneRate Cron] Invalid artwork ID found in start_counts result: ${row.res}`);
-      }
-    });
-    console.log(`[DoneRate Cron] Retrieved ${startResults.length} unique start counts from ClickHouse.`);
+    const clickhouseQueryStartTime = Date.now();
+    const aggregateResults = await clickhouseService.queryEvents<ClickHouseDoneRateAggregateResult>(aggregateQuery);
+    const clickhouseQueryElapsedMs = Date.now() - clickhouseQueryStartTime;
 
-    // 1.2 查询目标日期每个作品的独立完成用户数
-    const doneCountsQuery = `
-      SELECT
-          res,
-          count(DISTINCT uid) AS unique_dones
-      FROM ${CLICKHOUSE_EVENTS_TABLE}
-      WHERE event = 'color_done'
-        AND time >= toDateTime('${yesterdayStartString}')
-        AND time < toDateTime('${yesterdayEndString}')
-      GROUP BY res
-      HAVING res IS NOT NULL
-    `;
-    const doneResults = await clickhouseService.queryEvents<ClickHouseDoneCountResult>(doneCountsQuery);
+    const artworkStartCounts = new Map<string, number>();
     const artworkDoneCounts = new Map<string, number>();
-    doneResults.forEach((row) => {
+    const artworkTipCounts = new Map<string, number>();
+    let invalidArtworkRowCount = 0;
+
+    aggregateResults.forEach((row) => {
       if (row.res && mongoose.Types.ObjectId.isValid(row.res)) {
-        artworkDoneCounts.set(row.res, row.unique_dones);
+        artworkStartCounts.set(row.res, Number(row.unique_starts));
+        artworkDoneCounts.set(row.res, Number(row.unique_dones));
+        artworkTipCounts.set(row.res, Number(row.tip_count));
       } else {
-        console.warn(`[DoneRate Cron] Invalid artwork ID found in done_counts result: ${row.res}`);
+        invalidArtworkRowCount++;
+        console.warn(`[DoneRate Cron] Invalid artwork ID found in aggregate result: ${row.res}`);
       }
     });
-    console.log(`[DoneRate Cron] Retrieved ${doneResults.length} unique done counts from ClickHouse.`);
 
-    // 1.3 查询目标日期每个作品的使用道具数
-    const tipCountsQuery = `
-      SELECT
-          res,
-          count() AS tip_count
-      FROM ${CLICKHOUSE_EVENTS_TABLE}
-      WHERE event = 'color_tip'
-        AND time >= toDateTime('${yesterdayStartString}')
-        AND time < toDateTime('${yesterdayEndString}')
-      GROUP BY res
-      HAVING res IS NOT NULL
-    `;
-    const tipResults = await clickhouseService.queryEvents<ClickHouseTipCountResult>(tipCountsQuery);
-    const artworkTipCounts = new Map<string, number>();
-    tipResults.forEach((row) => {
-      if (row.res && mongoose.Types.ObjectId.isValid(row.res)) {
-        artworkTipCounts.set(row.res, row.tip_count);
-      } else {
-        console.warn(`[DoneRate Cron] Invalid artwork ID found in tip_counts result: ${row.res}`);
-      }
+    let totalStartUsers = 0;
+    let totalDoneUsers = 0;
+    let totalTipEvents = 0;
+    aggregateResults.forEach((row) => {
+      totalStartUsers += Number(row.unique_starts) || 0;
+      totalDoneUsers += Number(row.unique_dones) || 0;
+      totalTipEvents += Number(row.tip_count) || 0;
     });
-    console.log(`[DoneRate Cron] Retrieved ${tipResults.length} unique tip counts from ClickHouse.`);
+
+    console.log(
+      `[DoneRate Cron] ClickHouse aggregate query completed in ${clickhouseQueryElapsedMs}ms. Rows: ${aggregateResults.length}, valid artworks: ${artworkStartCounts.size}, invalid rows: ${invalidArtworkRowCount}, unique starts: ${totalStartUsers}, unique dones: ${totalDoneUsers}, tip events: ${totalTipEvents}.`
+    );
 
     // --- 2. 合并数据并更新 DoneRate 模型 (每日记录) ---
     let updatedRecordsCount = 0;