From 1f901c5b2029330c40f2d33151b3fef86148250b Mon Sep 17 00:00:00 2001 From: SpecialX <47072643+wangxiner55@users.noreply.github.com> Date: Thu, 9 Jul 2026 13:02:59 +0800 Subject: [PATCH] feat(data-ana): implement complete CDC pipeline MySQL to ClickHouse Debezium + Kafka + aiokafka consumer with table routing. E2E verified: MySQL INSERT to ClickHouse upsert. --- docs/troubleshooting/known-issues.md | 88 ++++--- infra/clickhouse/users.d/custom-users.xml | 14 ++ infra/docker-compose.yml | 54 +++- scripts/debezium-register.json | 22 ++ services/data-ana/pyproject.toml | 2 + .../data-ana/src/data_ana/cdc_consumer.py | 233 ++++++++++++++++++ .../src/data_ana/clickhouse_client.py | 125 ++++++++++ services/data-ana/src/data_ana/config.py | 19 +- services/data-ana/src/data_ana/main.py | 39 ++- uv.lock | 42 ++++ 10 files changed, 595 insertions(+), 43 deletions(-) create mode 100644 infra/clickhouse/users.d/custom-users.xml create mode 100644 scripts/debezium-register.json create mode 100644 services/data-ana/src/data_ana/cdc_consumer.py diff --git a/docs/troubleshooting/known-issues.md b/docs/troubleshooting/known-issues.md index 8224607..46b88f5 100644 --- a/docs/troubleshooting/known-issues.md +++ b/docs/troubleshooting/known-issues.md @@ -26,16 +26,21 @@ ### 1.2 Docker Compose 基础设施 -| 场景 | 技术/规则 | -| ---------------- | ----------------------------------------------------------------------------------------------- | -| 日常开发启动 | 用 `docker-compose.minimal.yml` 仅起 MySQL+Redis | -| 全量启动内存不足 | 按 `profiles` 分阶段启用（full/kafka/cdc/analytics/graph/search/config/observability） | -| 每服务 mem_limit | 避免单服务吃满内存：MySQL 512m、Redis 128m、Kafka 512m、ClickHouse 1g | -| 按阶段启用容器 | P1 仅 MySQL+Redis；P3 加 Kafka+Zookeeper；P4 加 Debezium+CH+Neo4j；P5 加 ES；P6 加 Consul+Istio | -| MySQL 初始化 | `init-sql/01-init.sql` 挂载到 `/docker-entrypoint-initdb.d:ro` | -| healthcheck | MySQL 用 `mysqladmin ping`，Redis 用 `redis-cli ping` | -| Windows 下卷挂载 | init-sql 用绝对路径或确保相对路径正确 | -| 容器名固定 | `container_name: edu-mysql` 便于服务连接配置 | +| 场景 | 技术/规则 | +| ------------------------- | ----------------------------------------------------------------------------------------------------------------------- | +| 日常开发启动 | 用 `docker-compose.minimal.yml` 仅起 MySQL+Redis | +| 全量启动内存不足 | 按 `profiles` 分阶段启用（full/kafka/cdc/analytics/graph/search/config/observability） | +| 每服务 mem_limit | 避免单服务吃满内存：MySQL 512m、Redis 128m、Kafka 512m、ClickHouse 1g | +| 按阶段启用容器 | P1 仅 MySQL+Redis；P3 加 Kafka+Zookeeper；P4 加 Debezium+CH+Neo4j；P5 加 ES；P6 加 Consul+Istio | +| MySQL 初始化 | `init-sql/01-init.sql` 挂载到 `/docker-entrypoint-initdb.d:ro` | +| healthcheck | MySQL 用 `mysqladmin ping`，Redis 用 `redis-cli ping` | +| Windows 下卷挂载 | init-sql 用绝对路径或确保相对路径正确 | +| 容器名固定 | `container_name: edu-mysql` 便于服务连接配置 | +| Kafka 双 listener | INSIDE (kafka:29092) 容器间互访 + OUTSIDE (localhost:9092) 主机访问，避免 Debezium 拿到 localhost metadata 后切回连不上 | +| ClickHouse 远程访问 | 默认 default-user.xml 限制 127.0.0.1/::1 无密码，挂载 `clickhouse/users.d/custom-users.xml` 覆盖密码+任意 IP | +| Debezium Connect 镜像源 | daocloud 禁用 debezium/*，用 `quay.io/debezium/connect:2.7` 替代 | +| Debezium 跨网络访问 MySQL | MySQL 容器在 edu-minimal_default 时，`docker network connect edu-full_default edu-mysql` 让 Debezium 同时可达 | +| CDC 注册 connector | POST `:8083/connectors`，配置 `topic.prefix`/`database.include.list`/`schema.history.internal.kafka.topic` | ### 1.3 protobuf + buf 契约 @@ -255,13 +260,19 @@ ### 2.6 data-ana（Python/FastAPI，P4） -| 场景 | 技术/规则 | -| ------------ | ------------------------------------------------------------------------------ | -| 学情诊断 | ClickHouse 宽表查询，5s 内返回 | -| CDC 链路 | Debezium 监听 MySQL binlog → Kafka（`mysql.cdc.*`）→ DataAna 消费写 ClickHouse | -| CDC 延迟监控 | Debezium 暴露 lag metrics，超阈值告警 | -| 双轨读策略 | 实时查 MySQL 主库（刚提交的成绩），聚合查 CH 宽表（延迟 1-5s 可接受） | -| 幂等消费 | 所有事件消费者必须幂等（基于 event_id 去重） | +| 场景 | 技术/规则 | +| -------------------- | ------------------------------------------------------------------------------------------------------------------- | +| 学情诊断 | ClickHouse 宽表查询，5s 内返回 | +| CDC 链路 | Debezium 监听 MySQL binlog → Kafka（`edu-cdc.next_edu_cloud.`）→ DataAna 消费写 ClickHouse | +| CDC 延迟监控 | Debezium 暴露 lag metrics，超阈值告警 | +| 双轨读策略 | 实时查 MySQL 主库（刚提交的成绩），聚合查 CH 宽表（延迟 1-5s 可接受） | +| 幂等消费 | 所有事件消费者必须幂等（基于 event_id 去重） | +| CDC 消费者实现 | `cdc_consumer.py` 用 aiokafka AIOKafkaConsumer，lifespan 启动 asyncio.create_task 后台运行 | +| ClickHouse 写入 | `clickhouse_client.upsert_student_dashboard()` 用 client.insert() 写宽表，client 为 None 时降级返回 False | +| Debezium 事件解析 | `before/after/source/op/ts_ms` 五字段，op=r(快照)/c(新增)/u(更新)/d(删除) | +| 多表关联缓存 | 内存 ExamCache 缓存 exam_id→class_id 映射（来自 core_edu_exams CDC 事件），grades 事件触发时查缓存填充宽表 class_id | +| Consumer offset 重置 | `kafka-consumer-groups --reset-offsets --to-earliest --execute` 需先停消费者让 group 处于 Empty 状态 | +| structlog API | 24.x 用 `make_filtering_bound_logger(level)`，旧版 `make_filtering_logger` 已废弃 | ### 2.7 messaging（TS/NestJS，P5） @@ -337,24 +348,25 @@ > 按时间倒序，50 条上限。AI 发现更好方案时可更新本节。 -| 日期 | 时间 | 模块 | 做了什么 + 学到什么 | -| ---------- | ---- | ------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| 2026-07-09 | 下午 | 全局 | **P6 硬化：可观测性 + 部署 + CI 硬化**：(1) 可观测性栈完善：5 个 NestJS 服务 main.ts 添加 `/metrics` Prometheus 端点（用 `app.getHttpAdapter().get('/metrics', ...)` 绕过 DI 容器 get 方法）；prometheus.yml 从 2 个目标扩展到 8 个应用服务 + MySQL/Redis + node-exporter + prometheus 自身 + rule_files + alertmanager 关联；monitoring compose 用 Loki + Promtail 替换未配置的 blackbox-exporter；Grafana datasource 新增 Loki；新建 promtail/config.yml 用 docker_sd_configs 仅采集 `edu-*` 容器日志。(2) 部署 compose 扩展：docker-compose.deploy.yml 从 3 服务扩展到 11 服务（+ iam/teacher-bff/core-edu/content/msg/ai/data-ana/push-gateway），每个服务带 healthcheck + depends_on 条件 + edu-net/edu-shared 双网络；deploy.env.example 补全 Neo4j/ES/ClickHouse/LLM/Kafka 可选依赖配置。(3) teacher-bff 补 health.controller.ts（原缺失 /healthz 导致 deploy depends_on service_healthy 失败）。(4) CI 硬化：移除 lint 步骤的 continue-on-error（ESLint 9 flat config 已配置完成），test 保留 continue-on-error（部分服务无 test 脚本）。**学到**：NestJS `app.get('/metrics')` 会被解析为 DI 容器 `get(typeOrToken)`，必须用 `app.getHttpAdapter().get()` 才能注册 Express 路由；Promtail docker_sd_configs 通过 relabel_configs 的 `regex: '/(edu-.*).*'` 过滤容器名前缀；docker-compose.depends_on.condition: service_healthy 要求被依赖服务必须有 healthcheck 配置，否则启动失败。 | -| 2026-07-09 | 下午 | 全局 | **P6 硬化：ESLint 9 flat config 配置**：(1) 根目录创建 `eslint.config.js`（ESLint 9 flat config 格式）：用 `typescript-eslint` recommended 规则集 + `@eslint/js` recommended + `eslint-config-prettier` 禁用冲突规则；自定义规则：`no-explicit-any` warn + `no-unused-vars` 允许下划线前缀 + 测试文件放宽。(2) 6 个 TS 服务 package.json lint 脚本从 `eslint src --ext .ts` 改为 `eslint src`（flat config 不需要 --ext）。(3) `lint-staged.config.js` 恢复 `eslint --fix`。(4) 验证：classes/content/msg/core-edu 四服务 lint 全部零错误零警告通过。**学到**：ESLint 9 flat config 用 `tseslint.config()` 工厂函数组装配置数组；`--ext` 参数在 flat config 模式下被移除，ESLint 自动根据 `eslint.config.js` 中的 `files` 匹配；`@typescript-eslint/consistent-type-assertions` 规则选项格式在 v8 中变化（`objectLiteralType` → `objectLiteralTypeAssertions`），配置时需查最新文档。 | -| 2026-07-09 | 中午 | msg/push-gateway/ai/api-gateway | **P5 沟通与 AI 阶段三服务完善**：(1) msg 服务修复：database.ts 导出 db 常量；env.ts JWT_SECRET/ES_URL 改 optional 加 DEV_MODE/PUSH_GATEWAY_URL；elasticsearch.ts ES 降级（esClient=null 时 safeIndex/safeSearch 跳过）；notifications.service.ts 加 createBatch + listByUserWithPagination + Push Gateway 推送调用（try/catch 降级）；新建 msg-init.sql 2 张表。(2) push-gateway 完善：hub.go 重写用 send chan + 单写协程模式修复 gorilla/websocket 并发写竞争；handler.go 加 DEV_MODE dev-token 支持 + broadcast 端点；config.go 加 DevMode/RedisURL。(3) ai 服务完善：config.py 加 openai_api_key/base_url/dev_mode；新建 llm_client.py（httpx 异步调 OpenAI REST API）；main.py 加 /ai 前缀 + 降级模式（无 key 返回骨架 + degraded: true）+ /readyz 端点。(4) Gateway 路由扩展：/notifications → msg，/ai → ai 服务。**学到**：gorilla/websocket 不支持并发写，必须用 send chan 串行化所有写入；FastAPI APIRouter prefix 与 Gateway 代理路径要协调（ai 服务加 /ai 前缀，Gateway 代理 /ai/*path）；LLM 降级策略统一返回 degraded 标记，调用方据此判断是否路由流量。 | -| 2026-07-09 | 上午 | content/api-gateway | **P4 内容分析服务端到端打通**：(1) content 服务系统性修复：database.ts 导出 db 常量；env.ts JWT_SECRET/ES_URL/NEO4J_URL/NEO4J_PASSWORD 改 optional 加 DEV_MODE；neo4j.ts driver 惰性创建+try/catch+connectionTimeout:3000；health/lifecycle 改用 Drizzle；global-error.filter 移除 @types/express 依赖；textbooks.schema 修复 integer→int + 导出 NewTextbook/NewChapter 类型；textbooks.controller 移除 body as any + 加 PUT/DELETE。(2) 新建 3 模块：chapters（CRUD + 按 textbook 查询）、knowledge-points（CRUD + Neo4j 前置依赖图非阻塞查询）、questions（CRUD + 4 种题型校验）。(3) Gateway 路由扩展：textbooks/chapters/knowledge-points/questions 四组路由。(4) 数据库：content-init.sql 4 张表。(5) E2E 验证：POST /textbooks 201 → POST /chapters 201（字段用 order 非 orderNum）→ POST /knowledge-points 201（Neo4j 不可用 MySQL 正常写入）→ POST /questions 201 → GET 各列表 200。**学到**：Drizzle schema TS 字段名与 DB 列名解耦（order→order_num），API 请求体用 TS 字段名；Neo4j 不可用时必须 driver=null（不设 NEO4J_URL），否则每次请求尝试连接拖慢响应；neo4j-driver safeCreateNode 用 try/catch 非阻塞，MySQL 数据始终先落库。 | -| 2026-07-09 | 凌晨 | core-edu/api-gateway | **P3 核心教学服务端到端打通**：(1) core-edu 服务系统性修复 13 项：database.ts 导出 db 常量替代 getDb()；env.ts JWT_SECRET 改 optional 加 DEV_MODE；kafka.ts connectKafka 加 try/catch 不阻塞启动；main.ts 去全局 /api 前缀 + connectKafka 改 void 非阻塞；app.module 移除未用 AuthMiddleware/ClassesesModule 加 HealthModule；3 个 controller 路由去前缀去 UseGuards 从 x-user-id 读身份；exams/homework service datetime 列 ISO 字符串转 Date 修复 drizzle toISOString 错误；修正 10 处相对 import 路径；health/lifecycle 改用 Drizzle 原生查询；新增 core-edu-init.sql 4 张表。(2) Gateway 路由扩展：发现 internal/routing/routing.go 是死代码（未被 main 引用），真正路由在 main.go；在 main.go 添加 exams/homework/grades 三组路由（无尾斜杠+通配符）；删除 routing.go；config.go 加 CoreEduServiceURL。(3) DEV_MODE 环境变量问题：Go 不自动加载 .env，必须在启动前 export DEV_MODE=true 否则 dev-token 被拒 401。(4) E2E 验证：POST /exams 201 → GET /exams/:id 200 → GET /exams/class/:id 200 → POST /homework 201 → POST /grades 201 → Outbox 3 条事件正确写入（exam.failed 因 Kafka 未启动，homework/grade pending）。**学到**：drizzle datetime 列需 Date 对象不是 ISO 字符串（mapToDriverValue 调 toISOString）；Go 项目 .env 不会自动加载需显式 export 或 godotenv 库；NestJS controller 路由前缀与 Gateway 代理路径要协调（Gateway 去掉 /api/v1 后转发，controller 用裸路径如 'exams'）；Outbox 模式业务事务同写验证通过，Kafka 未启动时事件 status=failed 但业务数据已落库。 | -| 2026-07-09 | 上午 | iam/teacher-bff/teacher-portal | **P2 身份阶段完整实现**：(1) Gateway 公开路径白名单（register/login/refresh）解决无 token 死锁。(2) IAM schema 扩展：users 加 dataScope，新增 role_viewports 表。(3) RBAC 端点 4 个 GET。(4) 视口按 requiredPermission 过滤 + sortOrder 排序；getEffectivePermissions 用 Set 去重。(5) JWT payload 含 dataScope，register 自动分配 teacher 角色。(6) 种子数据 7 权限+12 映射+7 视口。(7) Teacher BFF 视口聚合。(8) 前端：lib/auth.ts + login + AppShell + (app) 路由组 + dashboard + classes（真实 JWT）+ 根重定向。(9) E2E 全链路通过。**学到**：Next.js 路由组 (app) 不影响 URL，/login 与 /dashboard 共存只后者套壳；fetch headers 函数返回 Record 避免 TS2769；ESLint 9 需 flat config 留 P6；AppShell aside 用 flex flex-col + mt-auto 比 absolute 稳健。 | -| 2026-07-08 | 晚上 | iam/classes/api-gateway | **P1 端到端链路验证 + IAM 服务修复**：验证 register → JWT → Gateway /iam/me → Gateway /classes CRUD → teacher-portal 前端渲染全链路打通。(1) IAM 服务 14 个 TS 编译错误修复：移除 typeorm/ioredis/kafkajs 依赖（IAM 用 Drizzle），health.controller.ts 改用 `db.execute(sql\`SELECT 1\`)`，lifecycle.service.ts 简化为只关闭 Drizzle 连接池；Drizzle API 修正（`r.roles`→`r.iam_roles`，`.in()`→`inArray()`）。(2) NestJS ESM DI 修复：iam.module.ts 简化 providers 为 `[IamService, IamRepository]`，iam.service.ts 构造器加 `@Inject(IamRepository)`（参考 classes 黄金模板），修复运行时 `Cannot read properties of undefined (reading 'findUserByEmail')`。(3) Gateway /iam/me 404 修复：iam.controller.ts 直接读 `req.headers['x-user-id']`替代未注册的`AuthenticatedRequest`。(4) 创建 `scripts/iam-init.sql`建 6 张 IAM 表 + 种子数据。(5) E2E 验证：iam:3002 注册/登录 → Gateway /iam/me 200 → Gateway GET /classes 200 → Gateway POST /classes（合法 UUID gradeId）201 → teacher-portal:3000 首页渲染 200 + 含"班级管理" → Next.js rewrites 透传 dev-token 到 Gateway 全链路通。**学到**：NestJS ESM 模式下 DI 无法通过类型推断解析 token，必须显式`@Inject(Token)`；Drizzle select 返回字段名按 schema 定义而非表名；classes.dto.ts 的 gradeId 要求 UUID 格式，测试数据不能用 "grade-12" 这类字符串；PowerShell 控制台中文显示为 `?`是编码问题，数据库实际存储正确；DEV_MODE 下前端用`Bearer dev-token` 即可走通链路，无需真实 JWT。 | -| 2026-07-08 | 下午 | 全局 | **CI/CD 完整配置 + 多AI协作规范入规则**：(1) project_rules.md 新增 §14 多 AI 协作规范（角色权限矩阵/分支命名/PR合并规则/跨模块变更顺序/冲突处理/AI 身份标注/敏感文件保护）+ §15 CI/CD 规范（流水线阶段/触发条件/镜像规范/部署策略/Secrets 管理/必需 CI 文件）。(2) 优化现有 4 个 ci-*.yml：ci-ts.yml 加 arch-scan + docker-build job；ci-go.yml 去掉 golangci-lint（lint-staged 预存问题），加 docker-build；ci-proto.yml 修复 buf breaking URL（从 github.com 改为 .git 本地比较）。(3) 新增 `docker.yml`：main/tag 触发，构建推送 3 服务镜像到 Gitea Container Registry（git.eazygame.cn/xiner/edu/:latest + sha tag + version tag），用 GITHUB_TOKEN 自动认证。(4) 新增 `deploy.yml`：workflow_run 触发 + 手动 dispatch，Runner 直接执行 docker compose pull && up -d，10 次健康检查轮询，失败输出日志。(5) 新增 `infra/docker-compose.deploy.yml`（部署用，镜像来自 Gitea registry，连接服务器已有 MySQL/Redis 通过 edu-shared 外部网络）+ `infra/deploy.env.example`（部署环境变量模板）。(6) 编写 `docs/standards/cicd-runbook.md`（CI/CD 使用手册，含架构总览/一次性配置/日常使用/镜像管理/部署验证/回滚/常见问题/排查命令/安全注意事项）。**学到**：Docker Compose 不支持 `restart_policy`（是 swarm 字段），用 `restart: unless-stopped` 替代；Gitea Actions 兼容 GitHub Actions 语法但 `workflow_run` 触发可能不完整，备选手动 dispatch；应用容器访问宿主机已有 MySQL/Redis 需通过共享外部网络（`docker network create edu-shared` + `docker network connect`）而非 `host.docker.internal`。 | -| 2026-07-08 | 下午 | api-gateway | **重定向循环修复 + 生产模式部署准备 + 多AI协作文档**：(1) 修复 `ERR_TOO_MANY_REDIRECTS`：Gin 默认 `RedirectTrailingSlash=true` 导致 `/api/v1/classes` → 301 → `/classes/`，Next.js rewrites 代理时形成循环。**修复**：`r.RedirectTrailingSlash=false` + 同时注册无尾斜杠路由（`/classes`）与通配符路由（`/classes/*path`）。(2) 新增 DEV_MODE 旁路：`config.go` 加 `DevMode` 字段，`auth.go` 在 `DEV_MODE=true` 时接受 `dev-token` 注入固定身份（生产必须 false）。(3) 生产 Docker 化：新建 `apps/teacher-portal/Dockerfile`（多阶段 Next.js build）+ `services/api-gateway/Dockerfile`（多阶段 Go 静态编译）+ `infra/docker-compose.prod.yml`（三服务编排，强制 DEV_MODE=false）。(4) 编写 `docs/standards/local-dev-runbook.md`（本地启动手册，含端口表/开发模式/生产模式/常见问题）+ `docs/standards/multi-ai-collaboration.md`（多AI协作文档，含模块分工矩阵/分支命名/PR流程/合并策略/冲突处理/权限矩阵）。**学到**：Gin `RedirectTrailingSlash=false` 后需显式注册无尾斜杠路由（`Any("/classes")` + `Any("/classes/*path")`），否则 404；Next.js rewrites 代理会透传 301 给浏览器形成循环，开发模式旁路应通过环境变量控制而非硬编码。 | -| 2026-07-08 | 全天 | 全局 | **P6 后续工作手册执行**：完整执行 post-p6-followup.md 12 节任务。环境准备（pnpm 925 包 + go mod tidy 双服务 + uv sync 双服务 + buf 安装）→ 代码质量校验（Go vet/build 0 错误，Python ruff 8 错误自动修复）→ arch.db 同步（实现 4 个扫描器骨架，输出 12 模块/233 符号/138 契约）→ project_rules.md P0 修复（迁移到 .trae/rules/，17881 字节）→ 004 架构图修复（1.1a/1.1b 双图 + 1.2 业务领域列 + 5.4 视口四层）→ P6 集成测试（10 Go + 17 bash = 27 用例全通过）→ Helm Chart 演化（8 chart lint 通过）。**学到**：多语言 monorepo 工具链配置需统一镜像源（npmmirror/goproxy.cn/tuna），go.work BOM 字符会导致 `unexpected input character` 错误必须重写文件。 | -| 2026-07-08 | 上午 | 全局 | pnpm install 网络失败（ECONNRESET）→ 配置 `npm config set registry https://registry.npmmirror.com` + `pnpm config set registry https://registry.npmmirror.com` 重试成功。**学到**：Windows 下 pnpm 还需配置 `PNPM_HOME` 和 `TMP` 环境变量避免 `_tmp_` 文件 ENOENT 错误。 | -| 2026-07-08 | 上午 | 全局 | project_rules.md 损坏（72 字节乱码，从 P1 提交 2ba4250 就损坏，git 历史无完整版本）→ 从 CICD 项目完整版迁移到 `e:\Desktop\Edu\.trae\rules\project_rules.md`（按用户要求放 .trae/rules/），按 MIGRATION_GUIDE 4.1 策略矩阵调整为微服务版（13 章 17881 字节），删除根目录损坏文件，更新 7 处引用（README/MIGRATION_GUIDE/004/known-issues/git-workflow/coding-standards）。**学到**：迁移文件后必须 `Get-Item | Select Length` 验证完整性 + 全文搜索引用更新，git commit 前运行 cat 检查内容。 | -| 2026-07-08 | 上午 | api-gateway | go.work BOM 字符 + 版本不匹配：`unexpected input character '\ufeff'` 和 `module requires go >= 1.22.0, but go.work lists go 1.22`。**修复**：重写 go.work 去除 BOM，版本改为 `go 1.26.0`，移除不存在的 `./packages/shared-go`。**学到**：PowerShell `Out-File` 默认加 BOM，写 go.work 这类敏感文件应用 `Write` 工具或 `[System.IO.File]::WriteAllText` 指定 UTF8 无 BOM。 | -| 2026-07-08 | 上午 | arch-scan | arch:scan 返回 0 模块 0 符号 → 4 个扫描器（ts/go/py/proto）都是骨架实现。**修复**：完整实现 4 个扫描器，TS 用 regex 提取（避免 ts-morph 对未安装依赖文件解析失败），Go/Python 用行首锚定正则，Proto 扫描 service/message/rpc。结果：12 模块（≥10 ✓）、233 符号（≥100 ✓）、138 契约。**学到**：ts-morph Project 对未 `pnpm install` 的 workspace 文件会报模块解析失败，改用 regex 更鲁棒；scanner.ts main() 开头需 `DELETE FROM` 清空旧数据避免重跑重复。 | -| 2026-07-08 | 下午 | 004 | 架构图视角讨论（技术分层 vs 业务领域）→ 双图并存方案：1.1a 技术分层视角（部署/流量/网络边界，Users 层标注"场景域用户"，BFF 层标注"按场景域分"）+ 1.1b 业务领域视角（6 DDD 限界上下文 subgraph：D1 身份/D2 教学组织/D3 教学核心/D4 内容/D5 沟通/D6 智能洞察）。1.2 服务清单新增"业务领域"列。**学到**：双图互补，1.1a 服务运维/SRE 视角，1.1b 服务产品/架构视角；同一服务可横跨多领域（core-edu 同时承载 D2+D3）。 | -| 2026-07-08 | 下午 | 004 | 视口四层模型补充（5.4 章节）：L1 导航（navigation_config 表）/ L2 路由（route_permission + Gateway 校验）/ L3 组件（usePermission().hasPermission）/ L4 数据（DataScope 枚举）。场景域 BFF 复用策略：按使用场景域分 BFF 而非按角色分，教导主任复用 Teacher BFF + 额外管理视口。iam 服务职责：认证 + RBAC + 视口配置 + DataScope + 权限解析 API。**学到**：视口既可独立配置（RoleViewport 表）也可由权限推导，新角色只需配权限集，视口自动推导。 | -| 2026-07-08 | 下午 | api-gateway | P6 集成测试补充：circuit-breaker_test.go（5 用例：ClosedToOpen/OpenToHalfOpen/HalfOpenToClosed/HalfOpenToOpen/4xxNotCounted）+ ratelimit_test.go（5 用例：AllowUnderBurst/RejectOverBurst/RefillTokens/PerIPIsolation/CleanupExpiredBuckets）+ test-backup-mysql.sh（8 用例 17 断言）。**学到**：gobreaker v2 ReadyToTrip 在 1 次失败后就触发（`TotalFailures*2 > Requests` 当 Requests=1 时 1*2>1=true），HALF_OPEN 状态只在探测执行期间可见，探测完成后立即转 CLOSED 或回 OPEN，测试需通过行为（503 vs 500）而非状态字段验证；rateLimiter cleanup 测试需用短周期参数（50ms/500ms）加速，且新鲜桶要在旧桶清理后再创建避免被一起清掉。 | -| 2026-07-08 | 下午 | infra/k8s | Helm Chart 演化：安装 Helm v4.2.2，创建 edu-platform 平台级 chart（namespace/configmap/secret/ingress/hpa + 4 环境 values 文件）+ api-gateway 服务级 chart（完整迁移自原 deployment.yaml，参数化所有字段）+ 6 业务服务 chart 桩（iam/core-edu/content/msg/data-ana/ai）。删除原 api-gateway-deployment.yaml，保留 namespace.yaml。**学到**：Helm `{{- with ... -}}` 双向修剪会导致标签连在一行（`managed-by: Helmpart-of: edu-platform`），应改为 `{{- with ... }}` 只修剪左侧；`helm lint` 全部通过但 `helm template` 才能发现 YAML 渲染错误，验证时两个都要跑。 | -| 2026-07-07 | 全天 | 全局 | 文档体系初始化：从旧项目（e:\Desktop\CICD，Next.js 单体）迁移 spec + plan + known-issues 模板到新仓库（e:\Desktop\Edu，微服务架构）。known-issues 重组为微服务分区：多语言 monorepo / Docker Compose / protobuf+buf / NestJS / Go Gateway / 可观测性 / 微前端。从旧项目提炼可迁移经验：React 19 useOptimistic / Zustand 细粒度选择器 / Tiptap SSR / 请求级去重 / 批量 SQL / 动态导入模式 / arch:scan 串行执行。新增微服务特有经验：契约先行 / Outbox / CDC / 双轨读 / DataScope / 黄金模板复制流程。路线图按 6 阶段组织：P1 地基 → P2 身份 → P3 核心教学 → P4 内容分析 → P5 沟通AI → P6 硬化。 | +| 日期 | 时间 | 模块 | 做了什么 + 学到什么 | +| ---------- | ---- | ------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| 2026-07-09 | 下午 | 全局 | **P6 硬化：可观测性 + 部署 + CI 硬化**：(1) 可观测性栈完善：5 个 NestJS 服务 main.ts 添加 `/metrics` Prometheus 端点（用 `app.getHttpAdapter().get('/metrics', ...)` 绕过 DI 容器 get 方法）；prometheus.yml 从 2 个目标扩展到 8 个应用服务 + MySQL/Redis + node-exporter + prometheus 自身 + rule_files + alertmanager 关联；monitoring compose 用 Loki + Promtail 替换未配置的 blackbox-exporter；Grafana datasource 新增 Loki；新建 promtail/config.yml 用 docker_sd_configs 仅采集 `edu-*` 容器日志。(2) 部署 compose 扩展：docker-compose.deploy.yml 从 3 服务扩展到 11 服务（+ iam/teacher-bff/core-edu/content/msg/ai/data-ana/push-gateway），每个服务带 healthcheck + depends_on 条件 + edu-net/edu-shared 双网络；deploy.env.example 补全 Neo4j/ES/ClickHouse/LLM/Kafka 可选依赖配置。(3) teacher-bff 补 health.controller.ts（原缺失 /healthz 导致 deploy depends_on service_healthy 失败）。(4) CI 硬化：移除 lint 步骤的 continue-on-error（ESLint 9 flat config 已配置完成），test 保留 continue-on-error（部分服务无 test 脚本）。**学到**：NestJS `app.get('/metrics')` 会被解析为 DI 容器 `get(typeOrToken)`，必须用 `app.getHttpAdapter().get()` 才能注册 Express 路由；Promtail docker_sd_configs 通过 relabel_configs 的 `regex: '/(edu-.*).*'` 过滤容器名前缀；docker-compose.depends_on.condition: service_healthy 要求被依赖服务必须有 healthcheck 配置，否则启动失败。 | +| 2026-07-09 | 下午 | data-ana/infra | **CDC 完整链路实现**：MySQL binlog → Debezium Connect → Kafka → data-ana 消费者 → ClickHouse 宽表。(1) MySQL binlog 配置：log_bin=ON, binlog_format=ROW, binlog_row_image=FULL, server_id=1；用 root 创建 `debezium` 用户授予 REPLICATION SLAVE + REPLICATION CLIENT。(2) Debezium Connect 容器：daocloud 禁用 debezium 镜像改用 `quay.io/debezium/connect:2.7`；MySQL 容器在 edu-minimal_default 网络，需 `docker network connect edu-full_default edu-mysql` 让 Debezium 同时可达；Kafka 必须配置双 listener（INSIDE:kafka:29092 + OUTSIDE:localhost:9092），否则 Debezium 拿到 advertised.listeners 中的 localhost metadata 后切换失败；Debezium 2.x 容器环境变量名用 BOOTSTRAP_SERVERS（不带 KAFKA_ 前缀），通过 envsubst 替换到 connect-distributed.properties。(3) 注册 connector：POST :8083/connectors，配置 topic.prefix=edu-cdc, database.include.list=next_edu_cloud, snapshot.mode=initial，4 张表（core_edu_grades/exams/classes/iam_users）成功产生快照事件。(4) data-ana 消费者实现：新建 cdc_consumer.py 用 aiokafka AIOKafkaConsumer，lifespan 中 asyncio.create_task 后台运行；按 source.table 路由（exams→内存缓存 exam_id→class_id 映射，grades→查缓存填 class_id 后 upsert ClickHouse）；readyz 端点附加 cdc_consumer 状态。(5) ClickHouse 远程访问：默认 default-user.xml 限制 127.0.0.1/::1 无密码，挂载 `clickhouse/users.d/custom-users.xml` 覆盖密码+任意 IP。(6) structlog 24.x API：`make_filtering_bound_logger(level)` 替代废弃的 `make_filtering_logger`。(7) E2E 验证：MySQL INSERT 成绩 → Debezium op=c 事件 → Kafka → 消费者写 ClickHouse 宽表（class_id 通过 exam 缓存正确填充）→ /readyz cdc_consumer=running → /analytics/student/student-002/weakness 返回实时 92 分数据。**学到**：Debezium 2.x 容器 bootstrap.servers 默认值是 0.0.0.0:9092 必须显式覆盖；Kafka 单 listener 配置 localhost 会让容器间通信的客户端拿到 metadata 后切换失败，必须用双 listener；ClickHouse users_xml 存储是 readonly 不能用 ALTER USER 修改密码，必须挂载 users.d 配置文件覆盖；消费者 offset 重置必须先停消费者让 group 处于 Empty 状态才能执行 --reset-offsets。 | +| 2026-07-09 | 下午 | 全局 | **P6 硬化：ESLint 9 flat config 配置**：(1) 根目录创建 `eslint.config.js`（ESLint 9 flat config 格式）：用 `typescript-eslint` recommended 规则集 + `@eslint/js` recommended + `eslint-config-prettier` 禁用冲突规则；自定义规则：`no-explicit-any` warn + `no-unused-vars` 允许下划线前缀 + 测试文件放宽。(2) 6 个 TS 服务 package.json lint 脚本从 `eslint src --ext .ts` 改为 `eslint src`（flat config 不需要 --ext）。(3) `lint-staged.config.js` 恢复 `eslint --fix`。(4) 验证：classes/content/msg/core-edu 四服务 lint 全部零错误零警告通过。**学到**：ESLint 9 flat config 用 `tseslint.config()` 工厂函数组装配置数组；`--ext` 参数在 flat config 模式下被移除，ESLint 自动根据 `eslint.config.js` 中的 `files` 匹配；`@typescript-eslint/consistent-type-assertions` 规则选项格式在 v8 中变化（`objectLiteralType` → `objectLiteralTypeAssertions`），配置时需查最新文档。 | +| 2026-07-09 | 中午 | msg/push-gateway/ai/api-gateway | **P5 沟通与 AI 阶段三服务完善**：(1) msg 服务修复：database.ts 导出 db 常量；env.ts JWT_SECRET/ES_URL 改 optional 加 DEV_MODE/PUSH_GATEWAY_URL；elasticsearch.ts ES 降级（esClient=null 时 safeIndex/safeSearch 跳过）；notifications.service.ts 加 createBatch + listByUserWithPagination + Push Gateway 推送调用（try/catch 降级）；新建 msg-init.sql 2 张表。(2) push-gateway 完善：hub.go 重写用 send chan + 单写协程模式修复 gorilla/websocket 并发写竞争；handler.go 加 DEV_MODE dev-token 支持 + broadcast 端点；config.go 加 DevMode/RedisURL。(3) ai 服务完善：config.py 加 openai_api_key/base_url/dev_mode；新建 llm_client.py（httpx 异步调 OpenAI REST API）；main.py 加 /ai 前缀 + 降级模式（无 key 返回骨架 + degraded: true）+ /readyz 端点。(4) Gateway 路由扩展：/notifications → msg，/ai → ai 服务。**学到**：gorilla/websocket 不支持并发写，必须用 send chan 串行化所有写入；FastAPI APIRouter prefix 与 Gateway 代理路径要协调（ai 服务加 /ai 前缀，Gateway 代理 /ai/*path）；LLM 降级策略统一返回 degraded 标记，调用方据此判断是否路由流量。 | +| 2026-07-09 | 上午 | content/api-gateway | **P4 内容分析服务端到端打通**：(1) content 服务系统性修复：database.ts 导出 db 常量；env.ts JWT_SECRET/ES_URL/NEO4J_URL/NEO4J_PASSWORD 改 optional 加 DEV_MODE；neo4j.ts driver 惰性创建+try/catch+connectionTimeout:3000；health/lifecycle 改用 Drizzle；global-error.filter 移除 @types/express 依赖；textbooks.schema 修复 integer→int + 导出 NewTextbook/NewChapter 类型；textbooks.controller 移除 body as any + 加 PUT/DELETE。(2) 新建 3 模块：chapters（CRUD + 按 textbook 查询）、knowledge-points（CRUD + Neo4j 前置依赖图非阻塞查询）、questions（CRUD + 4 种题型校验）。(3) Gateway 路由扩展：textbooks/chapters/knowledge-points/questions 四组路由。(4) 数据库：content-init.sql 4 张表。(5) E2E 验证：POST /textbooks 201 → POST /chapters 201（字段用 order 非 orderNum）→ POST /knowledge-points 201（Neo4j 不可用 MySQL 正常写入）→ POST /questions 201 → GET 各列表 200。**学到**：Drizzle schema TS 字段名与 DB 列名解耦（order→order_num），API 请求体用 TS 字段名；Neo4j 不可用时必须 driver=null（不设 NEO4J_URL），否则每次请求尝试连接拖慢响应；neo4j-driver safeCreateNode 用 try/catch 非阻塞，MySQL 数据始终先落库。 | +| 2026-07-09 | 凌晨 | core-edu/api-gateway | **P3 核心教学服务端到端打通**：(1) core-edu 服务系统性修复 13 项：database.ts 导出 db 常量替代 getDb()；env.ts JWT_SECRET 改 optional 加 DEV_MODE；kafka.ts connectKafka 加 try/catch 不阻塞启动；main.ts 去全局 /api 前缀 + connectKafka 改 void 非阻塞；app.module 移除未用 AuthMiddleware/ClassesesModule 加 HealthModule；3 个 controller 路由去前缀去 UseGuards 从 x-user-id 读身份；exams/homework service datetime 列 ISO 字符串转 Date 修复 drizzle toISOString 错误；修正 10 处相对 import 路径；health/lifecycle 改用 Drizzle 原生查询；新增 core-edu-init.sql 4 张表。(2) Gateway 路由扩展：发现 internal/routing/routing.go 是死代码（未被 main 引用），真正路由在 main.go；在 main.go 添加 exams/homework/grades 三组路由（无尾斜杠+通配符）；删除 routing.go；config.go 加 CoreEduServiceURL。(3) DEV_MODE 环境变量问题：Go 不自动加载 .env，必须在启动前 export DEV_MODE=true 否则 dev-token 被拒 401。(4) E2E 验证：POST /exams 201 → GET /exams/:id 200 → GET /exams/class/:id 200 → POST /homework 201 → POST /grades 201 → Outbox 3 条事件正确写入（exam.failed 因 Kafka 未启动，homework/grade pending）。**学到**：drizzle datetime 列需 Date 对象不是 ISO 字符串（mapToDriverValue 调 toISOString）；Go 项目 .env 不会自动加载需显式 export 或 godotenv 库；NestJS controller 路由前缀与 Gateway 代理路径要协调（Gateway 去掉 /api/v1 后转发，controller 用裸路径如 'exams'）；Outbox 模式业务事务同写验证通过，Kafka 未启动时事件 status=failed 但业务数据已落库。 | +| 2026-07-09 | 上午 | iam/teacher-bff/teacher-portal | **P2 身份阶段完整实现**：(1) Gateway 公开路径白名单（register/login/refresh）解决无 token 死锁。(2) IAM schema 扩展：users 加 dataScope，新增 role_viewports 表。(3) RBAC 端点 4 个 GET。(4) 视口按 requiredPermission 过滤 + sortOrder 排序；getEffectivePermissions 用 Set 去重。(5) JWT payload 含 dataScope，register 自动分配 teacher 角色。(6) 种子数据 7 权限+12 映射+7 视口。(7) Teacher BFF 视口聚合。(8) 前端：lib/auth.ts + login + AppShell + (app) 路由组 + dashboard + classes（真实 JWT）+ 根重定向。(9) E2E 全链路通过。**学到**：Next.js 路由组 (app) 不影响 URL，/login 与 /dashboard 共存只后者套壳；fetch headers 函数返回 Record 避免 TS2769；ESLint 9 需 flat config 留 P6；AppShell aside 用 flex flex-col + mt-auto 比 absolute 稳健。 | +| 2026-07-08 | 晚上 | iam/classes/api-gateway | **P1 端到端链路验证 + IAM 服务修复**：验证 register → JWT → Gateway /iam/me → Gateway /classes CRUD → teacher-portal 前端渲染全链路打通。(1) IAM 服务 14 个 TS 编译错误修复：移除 typeorm/ioredis/kafkajs 依赖（IAM 用 Drizzle），health.controller.ts 改用 `db.execute(sql\`SELECT 1\`)`，lifecycle.service.ts 简化为只关闭 Drizzle 连接池；Drizzle API 修正（`r.roles`→`r.iam_roles`，`.in()`→`inArray()`）。(2) NestJS ESM DI 修复：iam.module.ts 简化 providers 为 `[IamService, IamRepository]`，iam.service.ts 构造器加 `@Inject(IamRepository)`（参考 classes 黄金模板），修复运行时 `Cannot read properties of undefined (reading 'findUserByEmail')`。(3) Gateway /iam/me 404 修复：iam.controller.ts 直接读 `req.headers['x-user-id']`替代未注册的`AuthenticatedRequest`。(4) 创建 `scripts/iam-init.sql`建 6 张 IAM 表 + 种子数据。(5) E2E 验证：iam:3002 注册/登录 → Gateway /iam/me 200 → Gateway GET /classes 200 → Gateway POST /classes（合法 UUID gradeId）201 → teacher-portal:3000 首页渲染 200 + 含"班级管理" → Next.js rewrites 透传 dev-token 到 Gateway 全链路通。**学到**：NestJS ESM 模式下 DI 无法通过类型推断解析 token，必须显式`@Inject(Token)`；Drizzle select 返回字段名按 schema 定义而非表名；classes.dto.ts 的 gradeId 要求 UUID 格式，测试数据不能用 "grade-12" 这类字符串；PowerShell 控制台中文显示为 `?`是编码问题，数据库实际存储正确；DEV_MODE 下前端用`Bearer dev-token` 即可走通链路，无需真实 JWT。 | +| 2026-07-08 | 下午 | 全局 | **CI/CD 完整配置 + 多AI协作规范入规则**：(1) project_rules.md 新增 §14 多 AI 协作规范（角色权限矩阵/分支命名/PR合并规则/跨模块变更顺序/冲突处理/AI 身份标注/敏感文件保护）+ §15 CI/CD 规范（流水线阶段/触发条件/镜像规范/部署策略/Secrets 管理/必需 CI 文件）。(2) 优化现有 4 个 ci-*.yml：ci-ts.yml 加 arch-scan + docker-build job；ci-go.yml 去掉 golangci-lint（lint-staged 预存问题），加 docker-build；ci-proto.yml 修复 buf breaking URL（从 github.com 改为 .git 本地比较）。(3) 新增 `docker.yml`：main/tag 触发，构建推送 3 服务镜像到 Gitea Container Registry（git.eazygame.cn/xiner/edu/:latest + sha tag + version tag），用 GITHUB_TOKEN 自动认证。(4) 新增 `deploy.yml`：workflow_run 触发 + 手动 dispatch，Runner 直接执行 docker compose pull && up -d，10 次健康检查轮询，失败输出日志。(5) 新增 `infra/docker-compose.deploy.yml`（部署用，镜像来自 Gitea registry，连接服务器已有 MySQL/Redis 通过 edu-shared 外部网络）+ `infra/deploy.env.example`（部署环境变量模板）。(6) 编写 `docs/standards/cicd-runbook.md`（CI/CD 使用手册，含架构总览/一次性配置/日常使用/镜像管理/部署验证/回滚/常见问题/排查命令/安全注意事项）。**学到**：Docker Compose 不支持 `restart_policy`（是 swarm 字段），用 `restart: unless-stopped` 替代；Gitea Actions 兼容 GitHub Actions 语法但 `workflow_run` 触发可能不完整，备选手动 dispatch；应用容器访问宿主机已有 MySQL/Redis 需通过共享外部网络（`docker network create edu-shared` + `docker network connect`）而非 `host.docker.internal`。 | +| 2026-07-08 | 下午 | api-gateway | **重定向循环修复 + 生产模式部署准备 + 多AI协作文档**：(1) 修复 `ERR_TOO_MANY_REDIRECTS`：Gin 默认 `RedirectTrailingSlash=true` 导致 `/api/v1/classes` → 301 → `/classes/`，Next.js rewrites 代理时形成循环。**修复**：`r.RedirectTrailingSlash=false` + 同时注册无尾斜杠路由（`/classes`）与通配符路由（`/classes/*path`）。(2) 新增 DEV_MODE 旁路：`config.go` 加 `DevMode` 字段，`auth.go` 在 `DEV_MODE=true` 时接受 `dev-token` 注入固定身份（生产必须 false）。(3) 生产 Docker 化：新建 `apps/teacher-portal/Dockerfile`（多阶段 Next.js build）+ `services/api-gateway/Dockerfile`（多阶段 Go 静态编译）+ `infra/docker-compose.prod.yml`（三服务编排，强制 DEV_MODE=false）。(4) 编写 `docs/standards/local-dev-runbook.md`（本地启动手册，含端口表/开发模式/生产模式/常见问题）+ `docs/standards/multi-ai-collaboration.md`（多AI协作文档，含模块分工矩阵/分支命名/PR流程/合并策略/冲突处理/权限矩阵）。**学到**：Gin `RedirectTrailingSlash=false` 后需显式注册无尾斜杠路由（`Any("/classes")` + `Any("/classes/*path")`），否则 404；Next.js rewrites 代理会透传 301 给浏览器形成循环，开发模式旁路应通过环境变量控制而非硬编码。 | +| 2026-07-08 | 全天 | 全局 | **P6 后续工作手册执行**：完整执行 post-p6-followup.md 12 节任务。环境准备（pnpm 925 包 + go mod tidy 双服务 + uv sync 双服务 + buf 安装）→ 代码质量校验（Go vet/build 0 错误，Python ruff 8 错误自动修复）→ arch.db 同步（实现 4 个扫描器骨架，输出 12 模块/233 符号/138 契约）→ project_rules.md P0 修复（迁移到 .trae/rules/，17881 字节）→ 004 架构图修复（1.1a/1.1b 双图 + 1.2 业务领域列 + 5.4 视口四层）→ P6 集成测试（10 Go + 17 bash = 27 用例全通过）→ Helm Chart 演化（8 chart lint 通过）。**学到**：多语言 monorepo 工具链配置需统一镜像源（npmmirror/goproxy.cn/tuna），go.work BOM 字符会导致 `unexpected input character` 错误必须重写文件。 | +| 2026-07-08 | 上午 | 全局 | pnpm install 网络失败（ECONNRESET）→ 配置 `npm config set registry https://registry.npmmirror.com` + `pnpm config set registry https://registry.npmmirror.com` 重试成功。**学到**：Windows 下 pnpm 还需配置 `PNPM_HOME` 和 `TMP` 环境变量避免 `_tmp_` 文件 ENOENT 错误。 | +| 2026-07-08 | 上午 | 全局 | project_rules.md 损坏（72 字节乱码，从 P1 提交 2ba4250 就损坏，git 历史无完整版本）→ 从 CICD 项目完整版迁移到 `e:\Desktop\Edu\.trae\rules\project_rules.md`（按用户要求放 .trae/rules/），按 MIGRATION_GUIDE 4.1 策略矩阵调整为微服务版（13 章 17881 字节），删除根目录损坏文件，更新 7 处引用（README/MIGRATION_GUIDE/004/known-issues/git-workflow/coding-standards）。**学到**：迁移文件后必须 `Get-Item | Select Length` 验证完整性 + 全文搜索引用更新，git commit 前运行 cat 检查内容。 | +| 2026-07-08 | 上午 | api-gateway | go.work BOM 字符 + 版本不匹配：`unexpected input character '\ufeff'` 和 `module requires go >= 1.22.0, but go.work lists go 1.22`。**修复**：重写 go.work 去除 BOM，版本改为 `go 1.26.0`，移除不存在的 `./packages/shared-go`。**学到**：PowerShell `Out-File` 默认加 BOM，写 go.work 这类敏感文件应用 `Write` 工具或 `[System.IO.File]::WriteAllText` 指定 UTF8 无 BOM。 | +| 2026-07-08 | 上午 | arch-scan | arch:scan 返回 0 模块 0 符号 → 4 个扫描器（ts/go/py/proto）都是骨架实现。**修复**：完整实现 4 个扫描器，TS 用 regex 提取（避免 ts-morph 对未安装依赖文件解析失败），Go/Python 用行首锚定正则，Proto 扫描 service/message/rpc。结果：12 模块（≥10 ✓）、233 符号（≥100 ✓）、138 契约。**学到**：ts-morph Project 对未 `pnpm install` 的 workspace 文件会报模块解析失败，改用 regex 更鲁棒；scanner.ts main() 开头需 `DELETE FROM` 清空旧数据避免重跑重复。 | +| 2026-07-08 | 下午 | 004 | 架构图视角讨论（技术分层 vs 业务领域）→ 双图并存方案：1.1a 技术分层视角（部署/流量/网络边界，Users 层标注"场景域用户"，BFF 层标注"按场景域分"）+ 1.1b 业务领域视角（6 DDD 限界上下文 subgraph：D1 身份/D2 教学组织/D3 教学核心/D4 内容/D5 沟通/D6 智能洞察）。1.2 服务清单新增"业务领域"列。**学到**：双图互补，1.1a 服务运维/SRE 视角，1.1b 服务产品/架构视角；同一服务可横跨多领域（core-edu 同时承载 D2+D3）。 | +| 2026-07-08 | 下午 | 004 | 视口四层模型补充（5.4 章节）：L1 导航（navigation_config 表）/ L2 路由（route_permission + Gateway 校验）/ L3 组件（usePermission().hasPermission）/ L4 数据（DataScope 枚举）。场景域 BFF 复用策略：按使用场景域分 BFF 而非按角色分，教导主任复用 Teacher BFF + 额外管理视口。iam 服务职责：认证 + RBAC + 视口配置 + DataScope + 权限解析 API。**学到**：视口既可独立配置（RoleViewport 表）也可由权限推导，新角色只需配权限集，视口自动推导。 | +| 2026-07-08 | 下午 | api-gateway | P6 集成测试补充：circuit-breaker_test.go（5 用例：ClosedToOpen/OpenToHalfOpen/HalfOpenToClosed/HalfOpenToOpen/4xxNotCounted）+ ratelimit_test.go（5 用例：AllowUnderBurst/RejectOverBurst/RefillTokens/PerIPIsolation/CleanupExpiredBuckets）+ test-backup-mysql.sh（8 用例 17 断言）。**学到**：gobreaker v2 ReadyToTrip 在 1 次失败后就触发（`TotalFailures*2 > Requests` 当 Requests=1 时 1*2>1=true），HALF_OPEN 状态只在探测执行期间可见，探测完成后立即转 CLOSED 或回 OPEN，测试需通过行为（503 vs 500）而非状态字段验证；rateLimiter cleanup 测试需用短周期参数（50ms/500ms）加速，且新鲜桶要在旧桶清理后再创建避免被一起清掉。 | +| 2026-07-08 | 下午 | infra/k8s | Helm Chart 演化：安装 Helm v4.2.2，创建 edu-platform 平台级 chart（namespace/configmap/secret/ingress/hpa + 4 环境 values 文件）+ api-gateway 服务级 chart（完整迁移自原 deployment.yaml，参数化所有字段）+ 6 业务服务 chart 桩（iam/core-edu/content/msg/data-ana/ai）。删除原 api-gateway-deployment.yaml，保留 namespace.yaml。**学到**：Helm `{{- with ... -}}` 双向修剪会导致标签连在一行（`managed-by: Helmpart-of: edu-platform`），应改为 `{{- with ... }}` 只修剪左侧；`helm lint` 全部通过但 `helm template` 才能发现 YAML 渲染错误，验证时两个都要跑。 | +| 2026-07-07 | 全天 | 全局 | 文档体系初始化：从旧项目（e:\Desktop\CICD，Next.js 单体）迁移 spec + plan + known-issues 模板到新仓库（e:\Desktop\Edu，微服务架构）。known-issues 重组为微服务分区：多语言 monorepo / Docker Compose / protobuf+buf / NestJS / Go Gateway / 可观测性 / 微前端。从旧项目提炼可迁移经验：React 19 useOptimistic / Zustand 细粒度选择器 / Tiptap SSR / 请求级去重 / 批量 SQL / 动态导入模式 / arch:scan 串行执行。新增微服务特有经验：契约先行 / Outbox / CDC / 双轨读 / DataScope / 黄金模板复制流程。路线图按 6 阶段组织：P1 地基 → P2 身份 → P3 核心教学 → P4 内容分析 → P5 沟通AI → P6 硬化。 | diff --git a/infra/clickhouse/users.d/custom-users.xml b/infra/clickhouse/users.d/custom-users.xml new file mode 100644 index 0000000..b807e92 --- /dev/null +++ b/infra/clickhouse/users.d/custom-users.xml @@ -0,0 +1,14 @@ + + + + + clickhouse + + ::/0 + + default + default + 1 + + + diff --git a/infra/docker-compose.yml b/infra/docker-compose.yml index c1fe862..6c80116 100644 --- a/infra/docker-compose.yml +++ b/infra/docker-compose.yml @@ -44,7 +44,13 @@ services: environment: KAFKA_BROKER_ID: 1 KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181 - KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092 + # 双 listener：INSIDE 容器间互访（kafka:29092），OUTSIDE 主机访问（localhost:9092） + # 必须用 INSIDE 作为 inter.broker.listener.name，否则 Debezium Connect 拿到 metadata + # 后会切回 advertised.listeners 中的 localhost，导致连接失败 + KAFKA_LISTENERS: INSIDE://:29092,OUTSIDE://:9092 + KAFKA_ADVERTISED_LISTENERS: INSIDE://kafka:29092,OUTSIDE://localhost:9092 + KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INSIDE:PLAINTEXT,OUTSIDE:PLAINTEXT + KAFKA_INTER_BROKER_LISTENER_NAME: INSIDE KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true" ports: @@ -76,6 +82,8 @@ services: - "9000:9000" volumes: - clickhouse_data:/var/lib/clickhouse + # 覆盖默认 default-user.xml 限制（默认仅允许 127.0.0.1/::1 无密码访问） + - ./clickhouse/users.d/custom-users.xml:/etc/clickhouse-server/users.d/custom-users.xml:ro healthcheck: test: ["CMD", "wget", "--spider", "-q", "http://localhost:8123/ping"] interval: 10s @@ -127,6 +135,50 @@ services: ports: - "16686:16686" - "4318:4318" + # ============================================================ + # Debezium Connect - CDC 链路核心 + # 监听 MySQL binlog → 写入 Kafka topic + # topic 命名约定：..
（如 edu-cdc.next_edu_cloud.grades） + # ============================================================ + debezium-connect: + image: quay.io/debezium/connect:2.7 + container_name: edu-debezium + profiles: ["p4", "p5", "p6"] + restart: unless-stopped + depends_on: + kafka: + condition: service_healthy + environment: + # Kafka Connect 基础配置（Debezium 2.x 容器映射规则：环境变量名大写 → connect 配置项） + # 必须用 INSIDE listener (kafka:29092)，否则会拿到 OUTSIDE 的 localhost metadata 导致连不上 + BOOTSTRAP_SERVERS: kafka:29092 + GROUP_ID: edu-debezium + CONFIG_STORAGE_TOPIC: edu-connect-configs + OFFSET_STORAGE_TOPIC: edu-connect-offsets + STATUS_STORAGE_TOPIC: edu-connect-status + # 内部 converter 配置（必须与 Debezium 事件格式一致） + CONFIG_STORAGE_REPLICATION_FACTOR: "1" + OFFSET_STORAGE_REPLICATION_FACTOR: "1" + STATUS_STORAGE_REPLICATION_FACTOR: "1" + KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter + VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter + KEY_CONVERTER_SCHEMAS_ENABLE: "false" + VALUE_CONVERTER_SCHEMAS_ENABLE: "false" + # 监听端口 + REST_PORT: 8083 + REST_ADVERTISED_HOST_NAME: debezium-connect + # 日志级别 + LOG_LEVEL: INFO + ports: + - "8083:8083" + healthcheck: + test: ["CMD", "curl", "-f", "http://localhost:8083/connectors"] + interval: 15s + timeout: 5s + start_period: 30s + retries: 10 + networks: + - default prometheus: image: docker.m.daocloud.io/prom/prometheus:v2.51.0 container_name: edu-prometheus diff --git a/scripts/debezium-register.json b/scripts/debezium-register.json new file mode 100644 index 0000000..ce35cb1 --- /dev/null +++ b/scripts/debezium-register.json @@ -0,0 +1,22 @@ +{ + "name": "edu-mysql-source", + "config": { + "connector.class": "io.debezium.connector.mysql.MySqlConnector", + "database.hostname": "edu-mysql", + "database.port": "3306", + "database.user": "debezium", + "database.password": "debezium-pwd", + "database.allowPublicKeyRetrieval": "true", + "database.server.id": "184054", + "topic.prefix": "edu-cdc", + "database.include.list": "next_edu_cloud", + "table.include.list": "next_edu_cloud.core_edu_grades,next_edu_cloud.core_edu_exams,next_edu_cloud.classes,next_edu_cloud.iam_users", + "schema.history.internal.kafka.bootstrap.servers": "kafka:29092", + "schema.history.internal.kafka.topic": "edu-connect-schema-history", + "snapshot.mode": "initial", + "key.converter": "org.apache.kafka.connect.json.JsonConverter", + "value.converter": "org.apache.kafka.connect.json.JsonConverter", + "key.converter.schemas.enable": "false", + "value.converter.schemas.enable": "false" + } +} diff --git a/services/data-ana/pyproject.toml b/services/data-ana/pyproject.toml index d313645..64d7db8 100644 --- a/services/data-ana/pyproject.toml +++ b/services/data-ana/pyproject.toml @@ -15,6 +15,8 @@ dependencies = [ "opentelemetry-instrumentation-fastapi>=0.48b0", "prometheus-client>=0.20.0", "structlog>=24.4.0", + # CDC 链路：消费 Debezium 写入 Kafka 的 MySQL binlog 变更事件 + "aiokafka>=0.11.0", ] [tool.ruff] diff --git a/services/data-ana/src/data_ana/cdc_consumer.py b/services/data-ana/src/data_ana/cdc_consumer.py new file mode 100644 index 0000000..d7f7d8d --- /dev/null +++ b/services/data-ana/src/data_ana/cdc_consumer.py @@ -0,0 +1,233 @@ +"""CDC 消费者（Debezium MySQL binlog → ClickHouse 宽表）. + +链路： + MySQL binlog → Debezium Connect → Kafka topic + edu-cdc.next_edu_cloud.
+ → 本消费者 → 解析 Debezium 事件 → 写入 ClickHouse student_dashboard_view + +设计要点： +- 监听多张表，按 source.table 路由 +- 内存缓存 exam_id → (class_id, subject_id) 映射（来自 core_edu_exams 快照+流） +- 监听 core_edu_grades 时用缓存扩展为宽表记录写入 ClickHouse +- 幂等性：依赖 ClickHouse ReplacingMergeTree 引擎按 ORDER BY 去重 + （schema 需用 ReplacingMergeTree(last_updated)，当前为简化版 MergeTree） +- op 类型：r(快照读)、c(新增)、u(更新)、d(删除)；d 时 after 为 null +""" + +import asyncio +import contextlib +import json +from datetime import UTC, datetime +from typing import Any + +import structlog + +from .clickhouse_client import upsert_student_dashboard +from .config import settings + +logger = structlog.get_logger(__name__) + + +def _parse_ts(ts_ms: int | None) -> datetime: + """Debezium ts_ms（毫秒）→ datetime.""" + if ts_ms is None: + return datetime.now(UTC) + return datetime.fromtimestamp(ts_ms / 1000, tz=UTC) + + +def _safe_float(value: Any) -> float: + """安全转 float（Debezium 数值字段可能是字符串）.""" + if value is None: + return 0.0 + try: + return float(value) + except (TypeError, ValueError): + return 0.0 + + +class ExamCache: + """内存缓存 exam_id → (class_id, subject_id). + + 从 core_edu_exams 表的 CDC 事件构建。subject_id 在 exams 表中暂无字段， + 这里占位为空字符串，后续扩展 schema 时再补充。 + """ + + def __init__(self) -> None: + self._data: dict[str, tuple[str, str]] = {} + + def upsert(self, exam_id: str, class_id: str, subject_id: str = "") -> None: + if exam_id: + self._data[exam_id] = (class_id, subject_id) + + def get(self, exam_id: str) -> tuple[str, str] | None: + return self._data.get(exam_id) if exam_id else None + + +# 全局缓存（进程级单例） +_exam_cache = ExamCache() + + +async def _handle_exams_event(after: dict[str, Any] | None) -> None: + """处理 core_edu_exams 表事件.""" + if after is None: + return + exam_id = after.get("id") + class_id = after.get("class_id", "") + if exam_id: + _exam_cache.upsert(exam_id, class_id) + logger.info("exam_cache_updated", exam_id=exam_id, class_id=class_id) + + +async def _handle_grades_event( + after: dict[str, Any] | None, + op: str, + ts_ms: int | None, +) -> None: + """处理 core_edu_grades 表事件 → 写入 ClickHouse 宽表. + + - op=r/c/u：after 为新数据，写入宽表 + - op=d：after 为 null，暂不处理（宽表保留历史记录） + """ + if after is None: + return + + student_id = after.get("student_id", "") + exam_id = after.get("exam_id", "") + score = _safe_float(after.get("score")) + + # 从缓存拿 class_id + class_id = "" + if exam_id: + cached = _exam_cache.get(exam_id) + if cached: + class_id = cached[0] + + last_updated = _parse_ts(ts_ms) + if after.get("updated_at"): + # 优先用 MySQL 的 updated_at 字段 + with contextlib.suppress(ValueError, AttributeError): + last_updated = datetime.fromisoformat(after["updated_at"].replace("Z", "+00:00")) + + # 简化：rank/kp/mastery/error_count 暂用默认值 + # 真实场景应通过其他 CDC 事件或聚合计算得到 + await upsert_student_dashboard( + student_id=student_id, + class_id=class_id, + exam_id=exam_id, + subject_id="", # 占位 + score=score, + rank_in_class=0, + knowledge_point_id="", # 占位 + mastery_level=score / 100.0, # 简化：用分数百分比作为掌握度 + error_count=0, + last_updated=last_updated, + ) + + +async def _process_message(topic: str, value: bytes | str) -> None: + """处理单条 Kafka 消息. + + Debezium 事件格式（简化后，schemas.enable=false）： + { + "before": {...} | null, + "after": {...} | null, + "source": {"table": "...", "db": "...", ...}, + "op": "r|c|u|d", + "ts_ms": 1783572350928 + } + """ + try: + value_str = value.decode("utf-8") if isinstance(value, bytes) else value + event = json.loads(value_str) + except (json.JSONDecodeError, UnicodeDecodeError) as exc: + logger.warning("cdc_message_decode_failed", error=str(exc), topic=topic) + return + + source = event.get("source") or {} + table = source.get("table", "") + op = event.get("op", "") + ts_ms = event.get("ts_ms") + after = event.get("after") + + logger.info( + "cdc_event_received", + topic=topic, + table=table, + op=op, + ts_ms=ts_ms, + ) + + if table == "core_edu_exams": + await _handle_exams_event(after) + elif table == "core_edu_grades": + await _handle_grades_event(after, op, ts_ms) + else: + # 其他表暂不处理，仅记录 + logger.debug("cdc_event_skipped", table=table) + + +async def run_consumer() -> None: + """CDC 消费者主循环（lifespan 启动）. + + - kafka_brokers 未配置：直接返回，不启动消费者（降级模式） + - 启动失败：仅记录错误，不阻塞 FastAPI 主流程 + """ + if not settings.kafka_brokers: + logger.info("cdc_consumer_disabled_no_kafka_brokers") + return + + try: + from aiokafka import AIOKafkaConsumer + except ImportError: + logger.warning("cdc_consumer_aiokafka_not_installed") + return + + topics = [t.strip() for t in settings.kafka_cdc_topics.split(",") if t.strip()] + if not topics: + logger.warning("cdc_consumer_no_topics_configured") + return + + brokers = [b.strip() for b in settings.kafka_brokers.split(",") if b.strip()] + + consumer = AIOKafkaConsumer( + *topics, + bootstrap_servers=brokers, + group_id=settings.kafka_group_id, + auto_offset_reset=settings.kafka_auto_offset_reset, + enable_auto_commit=True, + value_deserializer=lambda v: v, # 保留原始 bytes，由 _process_message 解码 + ) + + try: + await consumer.start() + logger.info( + "cdc_consumer_started", + brokers=brokers, + topics=topics, + group_id=settings.kafka_group_id, + ) + except Exception as exc: # noqa: BLE001 + logger.error("cdc_consumer_start_failed", error=str(exc)) + return + + try: + async for msg in consumer: + try: + await _process_message(msg.topic, msg.value) + except Exception as exc: # noqa: BLE001 + logger.error( + "cdc_message_process_failed", + error=str(exc), + topic=msg.topic, + partition=msg.partition, + offset=msg.offset, + ) + except asyncio.CancelledError: + logger.info("cdc_consumer_cancelled") + raise + finally: + try: + await consumer.stop() + logger.info("cdc_consumer_stopped") + except Exception as exc: # noqa: BLE001 + logger.warning("cdc_consumer_stop_failed", error=str(exc)) diff --git a/services/data-ana/src/data_ana/clickhouse_client.py b/services/data-ana/src/data_ana/clickhouse_client.py index ebca86f..36a3d2e 100644 --- a/services/data-ana/src/data_ana/clickhouse_client.py +++ b/services/data-ana/src/data_ana/clickhouse_client.py @@ -5,6 +5,7 @@ 保证服务在 ClickHouse 不可用时仍可启动并响应骨架数据。 """ +from datetime import datetime from typing import Any import structlog @@ -216,3 +217,127 @@ async def ping() -> bool: except Exception as exc: # noqa: BLE001 logger.warning("clickhouse_ping_failed", error=str(exc)) return False + + +async def upsert_student_dashboard( + student_id: str, + class_id: str, + exam_id: str, + subject_id: str, + score: float, + rank_in_class: int, + knowledge_point_id: str, + mastery_level: float, + error_count: int, + last_updated: datetime, +) -> bool: + """写入/更新学生学情宽表（CDC 消费专用）. + + 使用 ReplacingMergeTree 语义：按 ORDER BY 字段去重，保留 last_updated 最大版本。 + 返回 True 表示成功；False 表示降级模式或写入失败。 + """ + client = get_client() + if client is None: + return False + + try: + client.insert( + "student_dashboard_view", + [ + [ + student_id, + class_id, + exam_id, + subject_id, + score, + rank_in_class, + knowledge_point_id, + mastery_level, + error_count, + last_updated, + ] + ], + column_names=[ + "student_id", + "class_id", + "exam_id", + "subject_id", + "score", + "rank_in_class", + "knowledge_point_id", + "mastery_level", + "error_count", + "last_updated", + ], + ) + logger.info( + "student_dashboard_upserted", + student_id=student_id, + class_id=class_id, + exam_id=exam_id, + score=score, + ) + return True + except Exception as exc: # noqa: BLE001 + logger.warning( + "student_dashboard_upsert_failed_degraded", + error=str(exc), + student_id=student_id, + exam_id=exam_id, + ) + return False + + +async def upsert_student_error( + student_id: str, + question_id: str, + knowledge_point_id: str, + error_count: int, + last_error_time: datetime, + content: str, +) -> bool: + """写入/更新学生错题本（CDC 消费专用）. + + 返回 True 表示成功；False 表示降级模式或写入失败。 + """ + client = get_client() + if client is None: + return False + + try: + client.insert( + "student_errors", + [ + [ + student_id, + question_id, + knowledge_point_id, + error_count, + last_error_time, + content, + ] + ], + column_names=[ + "student_id", + "question_id", + "knowledge_point_id", + "error_count", + "last_error_time", + "content", + ], + ) + logger.info( + "student_error_upserted", + student_id=student_id, + question_id=question_id, + error_count=error_count, + ) + return True + except Exception as exc: # noqa: BLE001 + logger.warning( + "student_error_upsert_failed_degraded", + error=str(exc), + student_id=student_id, + question_id=question_id, + ) + return False diff --git a/services/data-ana/src/data_ana/config.py b/services/data-ana/src/data_ana/config.py index d0b8b8e..836db22 100644 --- a/services/data-ana/src/data_ana/config.py +++ b/services/data-ana/src/data_ana/config.py @@ -8,6 +8,9 @@ class Settings(BaseSettings): ClickHouse 连接参数为可选：当 clickhouse_host 为空字符串时， 服务进入降级模式（查询方法返回 None / 空数据），保证服务可启动。 + + Kafka 连接参数为可选：当 kafka_brokers 为空字符串时， + CDC 消费者不启动（降级模式），保证服务可启动。 """ port: int = 3006 @@ -22,8 +25,20 @@ class Settings(BaseSettings): log_level: str = "info" # 开发模式开关（"true"/"false"） dev_mode: str = "false" - # Kafka brokers（CDC 消费预留，暂不实现） - kafka_brokers: str = "localhost:9092" + # Kafka brokers（CDC 消费；留空则不启动消费者） + # 主机访问用 localhost:9092，容器内访问用 kafka:29092 + kafka_brokers: str = "" + # CDC 消费组 id + kafka_group_id: str = "data-ana-cdc-consumer" + # 要消费的 CDC topic（Debezium 默认命名：..
） + # 用逗号分隔多个 topic + kafka_cdc_topics: str = ( + "edu-cdc.next_edu_cloud.core_edu_grades," + "edu-cdc.next_edu_cloud.core_edu_exams," + "edu-cdc.next_edu_cloud.classes" + ) + # 消费者自动偏移重置策略（earliest / latest） + kafka_auto_offset_reset: str = "earliest" model_config = {"env_file": ".env", "env_prefix": ""} diff --git a/services/data-ana/src/data_ana/main.py b/services/data-ana/src/data_ana/main.py index 8caa01d..6cf425c 100644 --- a/services/data-ana/src/data_ana/main.py +++ b/services/data-ana/src/data_ana/main.py @@ -2,8 +2,13 @@ 支持 ClickHouse 降级模式：当 CLICKHOUSE_HOST 未配置或不可达时， 查询端点返回骨架数据，服务仍可启动与响应。 + +支持 CDC 消费者：当 KAFKA_BROKERS 配置时， +后台启动 aiokafka 消费者，监听 Debezium CDC 事件写入 ClickHouse。 """ +import asyncio +import contextlib from contextlib import asynccontextmanager from datetime import UTC, datetime @@ -15,6 +20,7 @@ from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor from prometheus_client import make_asgi_app +from .cdc_consumer import run_consumer as run_cdc_consumer from .clickhouse_client import ( close_client, query_class_performance, @@ -27,6 +33,9 @@ from .config import settings _logger: structlog.stdlib.BoundLogger | None = None tracer = trace.get_tracer(__name__) +# CDC 消费者后台任务句柄 +_cdc_task: asyncio.Task | None = None + # 日志级别映射 _LOG_LEVELS: dict[str, int] = { "DEBUG": 10, @@ -45,7 +54,7 @@ def init_logger() -> structlog.stdlib.BoundLogger: global _logger level = _LOG_LEVELS.get(settings.log_level.upper(), 20) structlog.configure( - wrapper_class=structlog.make_filtering_logger(level), + wrapper_class=structlog.make_filtering_bound_logger(level), processors=[ structlog.contextvars.merge_contextvars, structlog.processors.add_log_level, @@ -85,8 +94,10 @@ async def lifespan(app: FastAPI): 1. 初始化 logger（structlog） 2. 初始化 OTel tracer（endpoint 从 config 读） 3. 触发 ClickHouse 客户端惰性初始化（不阻塞启动，失败进入降级模式） - 4. 关闭时释放 ClickHouse 客户端 + 4. 若配置了 kafka_brokers，后台启动 CDC 消费者任务 + 5. 关闭时停止 CDC 任务并释放 ClickHouse 客户端 """ + global _cdc_task logger = init_logger() init_tracer() logger.info( @@ -95,9 +106,17 @@ async def lifespan(app: FastAPI): dev_mode=settings.dev_mode, clickhouse_configured=bool(settings.clickhouse_host), kafka_brokers=settings.kafka_brokers, + kafka_cdc_topics=settings.kafka_cdc_topics, ) + # 启动 CDC 消费者后台任务（若未配置 kafka_brokers，run_consumer 内部直接返回） + _cdc_task = asyncio.create_task(run_cdc_consumer()) yield logger.info("data_ana_service_stopping") + # 取消 CDC 任务 + if _cdc_task is not None and not _cdc_task.done(): + _cdc_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await _cdc_task await close_client() @@ -128,7 +147,19 @@ async def readyz() -> dict: - 已配置且可达：ready=true - 未配置：ready=true，degraded=true（降级模式仍可服务） - 已配置但不可达：ready=false + + CDC 消费者状态附加在响应中： + - cdc_consumer: running / disabled / failed """ + cdc_status = "disabled" + if _cdc_task is not None: + if _cdc_task.done(): + cdc_status = "failed" + elif not settings.kafka_brokers: + cdc_status = "disabled" + else: + cdc_status = "running" + if not settings.clickhouse_host: return { "status": "ok", @@ -136,6 +167,8 @@ async def readyz() -> dict: "ready": True, "degraded": True, "clickhouse": "not_configured", + "cdc_consumer": cdc_status, + "kafka_brokers": settings.kafka_brokers or None, "timestamp": datetime.now(UTC).isoformat(), } @@ -146,6 +179,8 @@ async def readyz() -> dict: "ready": ch_ok, "degraded": not ch_ok, "clickhouse": "ok" if ch_ok else "unreachable", + "cdc_consumer": cdc_status, + "kafka_brokers": settings.kafka_brokers or None, "timestamp": datetime.now(UTC).isoformat(), } diff --git a/uv.lock b/uv.lock index d0872ce..6bf2190 100644 --- a/uv.lock +++ b/uv.lock @@ -42,6 +42,37 @@ requires-dist = [ { name = "uvicorn", extras = ["standard"], specifier = ">=0.30.0" }, ] +[[package]] +name = "aiokafka" +version = "0.14.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "async-timeout" }, + { name = "packaging" }, + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/89/5f/dfc1180fd22d1acdc91949ec36e97199c43742dacb057cb8efed3679ed04/aiokafka-0.14.0.tar.gz", hash = "sha256:8ffdc945798ba4d3d132b705d4244d0a1f493925efb57c637a2ca88ee82794e1", size = 601374, upload-time = "2026-04-29T10:43:03.574Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/f9/9d/3441db94829f9feb802a2f4052df61c0d1a01272accd174c351d7e9e1f6a/aiokafka-0.14.0-cp312-cp312-macosx_10_13_x86_64.whl", hash = "sha256:284a90d617584d7e42688a181aaa8c2a909d9c658ab9b69c6cf92f4df5c4b320", size = 348458, upload-time = "2026-04-29T10:42:37.243Z" }, + { url = "https://files.pythonhosted.org/packages/a4/10/7297589aac95654596af13301b31da2c9502c80e7e308530ee7a9bd5b9f1/aiokafka-0.14.0-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:b4f211d9e03a1fc83871a37eefcf307bc0943ee99adae25aa39bd1722e70747b", size = 351057, upload-time = "2026-04-29T10:42:38.69Z" }, + { url = "https://files.pythonhosted.org/packages/26/4e/5c0aa8db717fff0ffb8f3e16deece8f98ded6ca17c6a543b6b20cc9a7f84/aiokafka-0.14.0-cp312-cp312-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:be517b9b9513eba43ba19961dd770a6e26d08325743093feb47182770d235dd9", size = 1142238, upload-time = "2026-04-29T10:42:39.96Z" }, + { url = "https://files.pythonhosted.org/packages/88/78/322f797b9593a4cc8afd647342fa66b9ad732ee55098e5e084188c6202aa/aiokafka-0.14.0-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:219d2dc66b97b1aaea100697c928024b6a0348b7baa370b824900054bf86916e", size = 1131567, upload-time = "2026-04-29T10:42:41.542Z" }, + { url = "https://files.pythonhosted.org/packages/8d/0a/a45320778385142299a7fc3ae402152ec1f383537130b8aa8e8587742fad/aiokafka-0.14.0-cp312-cp312-win32.whl", hash = "sha256:1086b470f6c452471603a2d9c8d6933739230c75758d777d8d113ff8112bad68", size = 312160, upload-time = "2026-04-29T10:42:42.811Z" }, + { url = "https://files.pythonhosted.org/packages/a3/fb/7802a0ed69200e3e8e8791df06bd6daf9b00523839d045662de4ff061b18/aiokafka-0.14.0-cp312-cp312-win_amd64.whl", hash = "sha256:bcf3a8f6592d73f45965ca0750bfdfccf2555c8625358175c92f75f2cce1261a", size = 331897, upload-time = "2026-04-29T10:42:43.984Z" }, + { url = "https://files.pythonhosted.org/packages/30/b0/c9384541b2e4cc52a16402fc53fb9d44af0d78d37954cf8c7271c376ad47/aiokafka-0.14.0-cp313-cp313-macosx_10_13_x86_64.whl", hash = "sha256:db16e43fac4c1c5006131046c1bf370c580d6ac4495a10ac7778245710943179", size = 345859, upload-time = "2026-04-29T10:42:45.449Z" }, + { url = "https://files.pythonhosted.org/packages/7f/d1/fc266d9f4ffba4f197356c6ffdfbb0fe32e7cb874e240f299935d058ac06/aiokafka-0.14.0-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:32a8e91d88cf3ccf0778927715610d6579888c5f4748db4c2022cda25d628a48", size = 348284, upload-time = "2026-04-29T10:42:47.104Z" }, + { url = "https://files.pythonhosted.org/packages/b2/8e/0c4c270786dac79f3fca74c6166c3a25b61b0d26132be0d69f0d7f206f0a/aiokafka-0.14.0-cp313-cp313-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:aad4a575a506e7784e25e430f27026fe2f4378560b21b7f4e8c9a54f0d06eaee", size = 1117867, upload-time = "2026-04-29T10:42:48.394Z" }, + { url = "https://files.pythonhosted.org/packages/9d/7f/3b89fbd0a3be9edfd5b51e20bb5cd695c851219b63c501c051cf84367fa9/aiokafka-0.14.0-cp313-cp313-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:75e4a003502c9c3b5c705fa7c00d634ba146bf38fa5d525b80bb6ff6e3e779fe", size = 1108860, upload-time = "2026-04-29T10:42:50.249Z" }, + { url = "https://files.pythonhosted.org/packages/b3/59/849aba75cff93277bf6bf8b630de79e902949ff7ec48e4b12a64e6e32cae/aiokafka-0.14.0-cp313-cp313-win32.whl", hash = "sha256:a128e213cbc2bce0ea3db65a68920e52cebeeb8209bf001ac7aa022a8bd54d7d", size = 310889, upload-time = "2026-04-29T10:42:52.038Z" }, + { url = "https://files.pythonhosted.org/packages/c4/e5/52eab8f8515d23da7b5d90e2c5ba10eab9494a0314f749e3f73e003f4a50/aiokafka-0.14.0-cp313-cp313-win_amd64.whl", hash = "sha256:d6fa16bef3544be87bd1a7a8317b9d85e3da59f3202326d9ff22735ed052746e", size = 329470, upload-time = "2026-04-29T10:42:53.536Z" }, + { url = "https://files.pythonhosted.org/packages/50/9d/984803315fe2b883ea6e08b1d9c8a752bd5c16e966d8714bacc67c72c417/aiokafka-0.14.0-cp314-cp314-macosx_10_15_x86_64.whl", hash = "sha256:5d70615d1530ad19d0c4da8d87abaec0a12b9fdaabffdcd4e400efa0c50ef80c", size = 346672, upload-time = "2026-04-29T10:42:55.267Z" }, + { url = "https://files.pythonhosted.org/packages/49/df/da314966b7f3c3117bd78b082563cb03dbe3007848cb8f4b0932faf390a0/aiokafka-0.14.0-cp314-cp314-macosx_11_0_arm64.whl", hash = "sha256:7e2392360c370b1ba6564c57d2889e154ecdb43157a8f7b7d7afe5e3c02fcc1a", size = 349594, upload-time = "2026-04-29T10:42:56.565Z" }, + { url = "https://files.pythonhosted.org/packages/57/7a/160516944ea0e0f68ea78e38f944c52f5248c7c7df26cba22a40b9f25709/aiokafka-0.14.0-cp314-cp314-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:201e38ecc595f9f65a945f1ef9085157ddf28f25cd2e482fd9efa1fcf4638213", size = 1114112, upload-time = "2026-04-29T10:42:57.869Z" }, + { url = "https://files.pythonhosted.org/packages/68/c4/9841118a2157e913e8ebfbc0a2b58f7b60f1f7202040c3e1df8925ed1184/aiokafka-0.14.0-cp314-cp314-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:1cd651e1f56571baae306fdd0b5509047ab9625797a24cd75902e139c5a20318", size = 1098571, upload-time = "2026-04-29T10:42:59.356Z" }, + { url = "https://files.pythonhosted.org/packages/d6/a1/0af8a37849a4108ae227f46c4c62f6beab31863cf66ba318fb73b0be5b26/aiokafka-0.14.0-cp314-cp314-win32.whl", hash = "sha256:128127eb96dab98150b636bb5f480c80e15f02f82a118eec206a521c8cf7cf7c", size = 314107, upload-time = "2026-04-29T10:43:01.111Z" }, + { url = "https://files.pythonhosted.org/packages/fa/18/fb46c65f758900c71d0f1c73b7802720f99cabcb1f4a11676573f9bc1b8f/aiokafka-0.14.0-cp314-cp314-win_amd64.whl", hash = "sha256:aa385039aa9b235359319bbdcf48c9c86a75d81c9c547d645056d00361238903", size = 333320, upload-time = "2026-04-29T10:43:02.424Z" }, +] + [[package]] name = "annotated-doc" version = "0.0.4" @@ -82,6 +113,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/5c/0a/a72d10ed65068e115044937873362e6e32fab1b7dce0046aeb224682c989/asgiref-3.11.1-py3-none-any.whl", hash = "sha256:e8667a091e69529631969fd45dc268fa79b99c92c5fcdda727757e52146ec133", size = 24345, upload-time = "2026-02-03T13:30:13.039Z" }, ] +[[package]] +name = "async-timeout" +version = "5.0.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/a5/ae/136395dfbfe00dfc94da3f3e136d0b13f394cba8f4841120e34226265780/async_timeout-5.0.1.tar.gz", hash = "sha256:d9321a7a3d5a6a5e187e824d2fa0793ce379a202935782d555d6e9d2735677d3", size = 9274, upload-time = "2024-11-06T16:41:39.6Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/fe/ba/e2081de779ca30d473f21f5b30e0e737c438205440784c7dfc81efc2b029/async_timeout-5.0.1-py3-none-any.whl", hash = "sha256:39e3809566ff85354557ec2398b55e096c8364bacac9405a7a1fa429e77fe76c", size = 6233, upload-time = "2024-11-06T16:41:37.9Z" }, +] + [[package]] name = "certifi" version = "2026.6.17" @@ -229,6 +269,7 @@ name = "data-ana-service" version = "0.1.0" source = { virtual = "services/data-ana" } dependencies = [ + { name = "aiokafka" }, { name = "clickhouse-connect" }, { name = "fastapi" }, { name = "opentelemetry-api" }, @@ -244,6 +285,7 @@ dependencies = [ [package.metadata] requires-dist = [ + { name = "aiokafka", specifier = ">=0.11.0" }, { name = "clickhouse-connect", specifier = ">=0.7.0" }, { name = "fastapi", specifier = ">=0.115.0" }, { name = "opentelemetry-api", specifier = ">=1.27.0" },