2 Commits

Author SHA1 Message Date
SpecialX
e9ea34fe53 feat(p6): production hardening with circuit breaker, backup, monitoring and chaos engineering
Some checks failed
CI Go / test (push) Has been cancelled
CI Python / test (push) Has been cancelled
CI TypeScript / test (push) Has been cancelled
CI Proto / lint (push) Failing after 8m7s
P6 生产硬化阶段交付物(46 文件):

## 1. API Gateway 中间件链(services/api-gateway/internal/middleware/)
- circuit-breaker.go: gobreaker v2 熔断器(5s 窗口/50% 错误率/30s OPEN→HALF_OPEN)
- ratelimit.go: 令牌桶限流(sync.Map + cleanup goroutine,默认 100rps/20 burst)
- cors.go: CORS 中间件(CORS_ORIGINS 环境变量)
- recovery.go: panic 恢复 + uuid request_id
- security.go: 安全头 + 请求体 10MB 限制
- requestid.go: 请求 ID 注入
- health/health.go: /healthz + /readyz 健康检查
- main.go: 重写注册全部中间件链(Recovery→RequestID→CORS→Security→BodyLimit→RateLimit→CircuitBreaker→Auth)

## 2. 基础设施硬化(infra/)
- backup/backup-mysql.sh: MySQL 全量备份(mysqldump+gzip,按服务独立)
- backup/restore-mysql.sh: 恢复脚本
- backup/backup-cron.sh: cron 调度入口(5 服务批量备份)
- alertmanager/alertmanager.yml: 告警路由(webhook + 邮件示例)
- prometheus/rules.yml: 8 条告警规则(服务可用性/性能/资源 3 组)
- grafana/dashboards/microservices-overview.json: 4 panel 仪表盘
- grafana/provisioning/: 数据源和仪表盘 provisioning
- k8s/namespace.yaml: 4 命名空间(edu-system/services/monitoring/ingress)
- k8s/api-gateway-deployment.yaml: Deployment + Service 骨架
- chaos/experiments.yaml: 3 个 Litmus 混沌实验(pod-kill/network-latency/disk-fill)
- docker-compose.monitoring.yml: 监控栈 profile
- security/secrets.example.env: 8 项密钥占位符
- security/waf-rules.conf: ModSecurity WAF 规则骨架

## 3. 业务服务健康检查 + 优雅停机(5 个 NestJS 服务)
- services/{iam,core-edu,content,msg,classes}/src/shared/health/: /healthz + /readyz
- services/{iam,core-edu,content,msg,classes}/src/shared/lifecycle/: OnModuleInit + OnApplicationShutdown

## 4. Python 服务健康检查
- services/{ai,data-ana}/src/health/health.py: FastAPI APIRouter

## 5. 运维文档
- docs/architecture/runbooks/p6-hardening.md: P6 总览 Runbook(9 章节)
- docs/architecture/runbooks/incident-response.md: 事件响应手册(5 章节)
- docs/architecture/004-p6-addendum.md: 004 架构补记 P6 章节
- docs/troubleshooting/known-issues-p6-addendum.md: 15 条 P6 场景→技术映射

## 验收信号
- RPO ≤ 15min(MySQL 备份 + binlog PITR)
- RTO ≤ 30min(K8s 滚动更新 + DNS 切换)
- P99 ≤ 500ms(熔断 + 限流 + 缓存)
- 熔断器错误率 > 50% 触发 OPEN
- 限流 100rps/20 burst
- 备份保留 7 天
- 混沌实验每月 1 次
2026-07-08 02:16:58 +08:00
SpecialX
7474a92e3b feat(p5): messaging, push gateway and AI assistant services
P5 阶段交付物:
- services/msg: 消息通知服务(NestJS)
  - notifications: 发送通知 + ES 全文检索 + search
  - config/elasticsearch.ts: ES Client 单例
  - package.json: 补充 @opentelemetry/sdk-node + exporter-trace-otlp-http
- services/push-gateway: WebSocket 推送网关(Go Gin)
  - internal/hub/hub.go: WebSocket 连接池管理(Register/Unregister/SendToUser)
  - internal/ws/handler.go: JWT 鉴权 + WebSocket 升级 + 内部推送 API
- services/ai: AI 辅助服务(Python FastAPI)
  - /chat + /chat/stream(SSE 流式)
  - /generate/question + /optimize/expression
  - config.py: OpenAI 兼容 API 配置
- packages/shared-proto/proto/msg.proto: NotificationService 契约(send/search)
- packages/shared-proto/proto/ai.proto: AiService 契约(含 stream 方法)
2026-07-08 01:39:02 +08:00
80 changed files with 4493 additions and 3 deletions

View File

@@ -0,0 +1,107 @@
## 15. P6 生产硬化(补记)
> 本章为 `004_architecture_impact_map.md` 的 P6 阶段补记,记录生产硬化引入的横切关注点与基础设施栈。
> 维护规则与正文一致:源码变更后同步 `npm run arch:scan` 更新 arch.db。
### 15.1 横切关注点矩阵
| 关注点 | NestJS 服务 | Python 服务 | Go 网关 | 实现位置 |
|--------|-------------|-------------|---------|----------|
| 健康检查 | HealthController | health.py | /healthz | shared/health, src/health |
| 优雅停机 | LifecycleService | FastAPI lifespan | enableShutdownHooks | shared/lifecycle |
| 熔断 | 经 Gateway | 经 Gateway | gobreaker v2 | api-gateway/middleware |
| 限流 | 经 Gateway | 经 Gateway | token bucket | api-gateway/middleware |
| 链路追踪 | tracer.ts | OpenTelemetry | OpenTelemetry | shared/observability |
| 指标 | metrics.ts | prometheus_fastapi | prometheus | shared/observability |
| 日志 | logger.ts | structlog | zap | shared/observability |
| 错误处理 | global-error.filter | exception handler | middleware | shared/errors |
### 15.2 API Gateway 中间件链
请求流经顺序(出向到下游服务):
```
请求入口
→ WAF规则匹配
→ CORS
→ 限流token bucket按 route+tenant
→ 熔断gobreaker v2按下游服务
→ 重试(指数退避,仅幂等)
→ 链路追踪注入
→ 转发到下游
→ 响应 → 指标记录 → 返回
```
### 15.3 可观测性栈
```
应用层NestJS / Python / Go
→ OpenTelemetry SDKtrace + metrics
→ OTLP exporter
→ 采集层
├─ Prometheusmetrics
├─ Tempo / Jaegertrace
└─ Loki / ELKlog
→ 展示层
├─ Grafana仪表盘
└─ Alertmanager告警路由
```
关键指标命名约定:
- `http_request_duration_seconds`histogram含 service/route/status 维度)
- `circuit_breaker_state`gauge0=Closed / 1=Open / 2=HalfOpen
- `rate_limiter_rejected_total`counter
- `db_connections_in_use`gauge
- `kafka_consumer_lag`gauge
### 15.4 安全栈
| 层 | 机制 | 配置位置 |
|----|------|----------|
| 边缘 | WAF + DDoS 防护 | Cloudflare / 入口 LB |
| 网关 | JWT 校验 + 限流 + CORS | api-gateway |
| 服务 | requirePermission 权限点 | modules/*/actions |
| 数据 | 字段加密 + 审计日志 | data-access |
| 密钥 | KMS + K8s Secret + 轮换 | deploy/k8s/secrets |
| 传输 | mTLS服务间可选+ TLS边缘 | mesh / ingress |
### 15.5 健康检查约定
- `GET /healthz`liveness仅返回进程存活不检查依赖避免滚动重启雪崩
- `GET /readyz`readiness检查 DB 等关键依赖,失败返回 503
- K8s 探针livenessProbe → /healthzreadinessProbe → /readyz
- Python 服务 readyz 简化为 ok + TODO待依赖客户端就绪后补全
- 无需鉴权,必须在路由白名单中放行
### 15.6 优雅停机约定
- NestJS`app.enableShutdownHooks()` 注册 SIGTERM/SIGINT 钩子
- LifecycleService 实现 OnApplicationShutdown按序关闭Kafka producer → Redis → DataSource
- K8s`terminationGracePeriodSeconds=60`preStop hook 可加 sleep 5s 摘流量
- PythonFastAPI lifespan shutdown 事件,关闭连接池
- 销毁顺序理由:先停外部消息生产(避免新事件),再关缓存,最后关 DB
### 15.7 灾难恢复策略
- 备份CronJob 每 15minPostgreSQL + Redis + Kafka offset
- 恢复:`scripts/restore/`,月度演练验证 RTO
- 多 AZPod 反亲和 + DB 同步复制 + Redis 哨兵 + Kafka ISR=2
- DNS 切换区域级故障TTL=60s季度演练
### 15.8 与正文章节的对应
| 本章小节 | 对应正文章节 |
|----------|--------------|
| 横切关注点 | 第 3 章 共享内核 |
| 中间件链 | 第 5 章 API Gateway |
| 可观测性 | 第 10 章 可观测性 |
| 安全栈 | 第 11 章 安全 |
| 健康检查 | 第 6 章 服务边界 |
| 灾难恢复 | 第 12 章 部署与运维 |
### 15.9 同步要求
新增导出符号需在落地到 Edu 仓库后运行 `npm run arch:scan` 更新 arch.db
- `HealthController``HealthModule`5 个 NestJS 服务)
- `LifecycleService`5 个 NestJS 服务)
- `health.py` routerai、data-ana

View File

@@ -0,0 +1,171 @@
# 事件响应手册
> 目标:规范生产事件的分级、响应、处置与复盘,确保 RTO ≤ 30min
> 关联:`docs/architecture/runbooks/p6-hardening.md`
## 1. 事件分级
| 级别 | 定义 | 影响 | 响应时效 | 升级 |
|------|------|------|----------|------|
| P0 | 全站不可用 / 核心数据损坏 | 全部用户 | 5 分钟内响应 | 立即升级至 CTO |
| P1 | 核心功能不可用 / 关键 SLO 破坏 | 大量用户 | 10 分钟内响应 | 升级至服务负责人 |
| P2 | 部分功能降级 / 非核心故障 | 部分用户 | 30 分钟内响应 | 服务负责人跟进 |
| P3 | 单点告警 / 潜在风险 | 少量/无用户 | 工作时间内响应 | On-Call 自行处理 |
### 1.1 分级示例
- P0网关全挂、主 DB 不可用且无法故障转移、数据丢失超过 RPO
- P1登录不可用、课程播放不可用、Kafka 生产阻塞
- P2消息推送延迟、报表生成失败、单个非核心服务宕机
- P3单 Pod 重启、磁盘使用率告警、慢查询告警
## 2. 响应流程
```
发现 → 确认 → 升级 → 处理 → 恢复 → 复盘
```
### 2.1 发现
- 告警来源Prometheus / Alertmanager / 用户反馈 / 人工巡检
- 第一动作:在 On-Call 群贴告警,标注收到时间
### 2.2 确认
- On-Call 5 分钟内确认告警真实性
- 排除误报(探针抖动、已知维护窗口)
- 初步定级创建事故工单Jira / 飞书项目)
### 2.3 升级
- 超出处置能力 → 立即升级
- P0/P1 → 拉事故群,通知相关服务 Owner
- 涉及外部公告 → 通知客服与公关
### 2.4 处理
- 遵循"先恢复,后定位"原则
- 优先使用预案:回滚、扩容、降级、熔断、切换
- 每个操作记录时间戳与执行人
- 关键决策需事故指挥确认
### 2.5 恢复
- 健康检查通过、SLO 恢复
- 观察 15 分钟确认稳定
- 关闭事故工单,进入复盘
### 2.6 复盘
- 48 小时内提交复盘报告
- 无 blame 文化:对事不对人
- 输出改进项,录入 `docs/architecture/roadmap/tech-debt.md`
## 3. On-Call 轮值
### 3.1 轮值制度
- 主备双人轮值,每周轮换
- 工作日9:00-21:00 主,其余备
- 节假日:全天主备
- 交接:周一 10:00 站会交接,遗留问题清单
### 3.2 联络方式
- 电话:主 + 备 + 升级链
- 即时通讯:飞书 On-Call 群
- 告警PagerDuty / 飞书机器人
### 3.3 响应要求
- P0/P1电话 5 分钟内接听
- 告警确认:群内 5 分钟内回复
- 无法响应:自动升级至备值
## 4. 沟通模板
### 4.1 事故通报(初报)
```
【事故通报】<P级别> - <简述>
时间:<YYYY-MM-DD HH:MM>
级别:<P0/P1/P2/P3>
影响:<受影响功能/用户范围>
现状:<已知信息>
负责人:<On-Call>
下一步:<计划动作>
```
### 4.2 进展更新(每 30 分钟或重大变化)
```
【进展更新】<事故标题>
时间:<HH:MM>
进展:<自上次以来发生/完成的事>
当前状态:<仍受影响的功能>
下一步:<接下来 30 分钟计划>
```
### 4.3 恢复通知
```
【恢复通知】<事故标题>
恢复时间:<HH:MM>
持续时长:<时长>
原因:<根因摘要>
影响:<最终影响评估>
后续:<复盘会时间>
```
### 4.4 复盘报告
```
【复盘报告】<事故标题>
时间:<起止时间>
级别:<P级别>
影响:<用户/功能/数据>
时间线:<关键事件时间轴>
根因:<5why 分析>
处置:<做了什么、有效/无效>
改进项:<TODO + 负责人 + 截止>
经验:<可沉淀到 known-issues / runbook 的内容>
```
## 5. 常见事故处置
### 5.1 DB 主从切换
1. 确认主库故障(健康检查、连接超时)
2. 触发自动故障转移Patroni / Orchestrator
3. 若自动失败,手动 `./scripts/db/failover.sh --service <svc>`
4. 更新连接配置 / 刷新连接池
5. 验证读写正常、数据位点
6. 旧主恢复后作为从库加入
### 5.2 Kafka 消费堆积
1. 查看堆积:`kubectl exec -- kafka-consumer-lag`
2. 定位慢消费者查日志、trace
3. 扩容消费者副本HPA 或手动)
4. 若处理逻辑慢:临时降级非核心处理
5. 堆积消化后恢复
6. 复盘:扩容阈值、消费者并发配置
### 5.3 服务雪崩
1. 确认雪崩源头(哪个下游故障)
2. 确认熔断器已打开Gateway 状态)
3. 若未打开:手动 `./scripts/gateway/circuit-breaker.sh open <service>`
4. 降级非核心功能
5. 扩容上游服务应对重试流量
6. 修复下游、半开试探、逐步恢复
7. 复盘:熔断参数、依赖隔离
## 附录:升级链
| 级别 | 第一响应 | 升级 1 | 升级 2 | 升级 3 |
|------|----------|--------|--------|--------|
| P0 | On-Call | 服务负责人 | 架构负责人 | CTO |
| P1 | On-Call | 服务负责人 | 架构负责人 | - |
| P2 | On-Call | 服务负责人 | - | - |
| P3 | On-Call | - | - | - |

View File

@@ -0,0 +1,264 @@
# P6 生产硬化 Runbook
> 阶段P6 生产硬化
> 目标指标RPO ≤ 15minRTO ≤ 30minP99 ≤ 500ms
> 维护者SRE 团队
> 关联文档:`docs/architecture/004_architecture_impact_map.md`、`docs/architecture/roadmap/tech-debt.md`、`docs/architecture/runbooks/incident-response.md`
## 1. 概览
P6 阶段围绕"稳定、可观测、可恢复"三大主题,对业务服务进行生产硬化,确保系统在流量峰值、依赖故障、区域级灾难下仍能满足核心 SLO。
### 1.1 目标指标
| 指标 | 目标 | 度量来源 |
|------|------|----------|
| RPO恢复点目标 | ≤ 15 分钟 | 备份调度日志 + WAL 位点 |
| RTO恢复时间目标 | ≤ 30 分钟 | 故障注入演练计时 |
| P99 延迟 | ≤ 500 ms | Prometheus http_request_duration_seconds |
| 可用性 | ≥ 99.9% | 多区域健康探针汇总 |
| 错误率 | ≤ 0.1% | Gateway 5xx 比率 |
### 1.2 交付物清单
- API Gateway 熔断/限流中间件Gogobreaker v2 + token bucket
- 备份与恢复脚本(`scripts/backup/``scripts/restore/`
- Prometheus 告警规则 + Alertmanager 路由配置
- Grafana 仪表盘服务总览、SLO、依赖健康
- 混沌工程实验库(`chaos/`
- K8s 部署 manifest`deploy/k8s/`
- 健康检查标准化(`/healthz``/readyz`
- 优雅停机(`lifecycle.service.ts` + `app.enableShutdownHooks()`
## 2. 熔断与限流
### 2.1 中间件使用
API GatewayGo在路由链路上统一接入
- 熔断器:`github.com/sony/gobreaker/v2`,按下游服务维度建桶
- 限流:基于 `sync.Map` 的令牌桶,按 `route + tenant` 维度限流
熔断器配置示例(伪代码):
```go
cb := gobreaker.NewCircuitBreaker[any](gobreaker.Settings{
Name: "core-edu",
MaxRequests: 5, // 半开态最大试探请求
Interval: 60 * time.Second, // 计数窗口
Timeout: 30 * time.Second, // 开启态冷却
ReadyToTrip: func(c gobreaker.Counts) bool {
return c.ConsecutiveFailures > 5 || c.TotalFailures/c.TotalRequests > 0.5
},
})
```
### 2.2 参数调优
| 参数 | 默认 | 推荐范围 | 调优依据 |
|------|------|----------|----------|
| MaxRequests半开试探 | 5 | 3-10 | 下游恢复速度 |
| Timeout冷却 | 30s | 10-60s | 下游故障平均恢复时间 |
| ConsecutiveFailures | 5 | 3-10 | 误报容忍度 |
| 限流 QPS租户级 | 1000 | 500-5000 | 租户 SLA 等级 |
| 桶清理周期 | 60s | 30-120s | 内存占用与精度权衡 |
### 2.3 故障演练流程
1. 在预发环境注入下游延迟tc/netem 500ms+
2. 观察熔断器状态切换Closed → Open → Half-Open → Closed
3. 验证限流桶在窗口边界正确清理sync.Map + ticker
4. 记录 P99 与错误率到演练报告
5. 回滚注入:`./chaos/rollback.sh <experiment>`
## 3. 备份与恢复
### 3.1 备份脚本
- 位置:`scripts/backup/backup-cron.sh`
- 调度K8s CronJob每 15 分钟一次(满足 RPO ≤ 15min
- 内容PostgreSQL 全量 + WAL 归档 + Redis RDB + Kafka topic offset 快照
- 产物写入对象存储MinIO/S3保留策略 7d日备 30d
执行:
```bash
./scripts/backup/backup-cron.sh --service iam --type full
./scripts/backup/backup-cron.sh --service iam --type incr
```
### 3.2 恢复演练流程
1. 在隔离环境拉起空集群
2. 执行 `./scripts/restore/restore.sh --service <svc> --backup-id <id>`
3. 校验数据一致性(行数、最新位点、校验和)
4. 记录恢复耗时,验证 RTO ≤ 30min
5. 演练频率:每月一次
### 3.3 RPO 验证方法
- 对比最近一次备份时间与当前时间差
- 查询 `backup_log``last_success_at` 字段
- 告警:连续 2 个周期30min未成功备份 → P1
## 4. 监控告警
### 4.1 Prometheus 规则
位置:`deploy/observability/prometheus/rules/`
关键规则:
- `HighErrorRate`5xx 比率 > 1% 持续 5min
- `HighLatencyP99`P99 > 500ms 持续 5min
- `CircuitBreakerOpen`:熔断器处于 Open 态 > 1min
- `BackupStale`:备份超过 30min 未成功
- `DBConnectionsExhausted`:连接池使用率 > 90%
- `KafkaConsumerLag`:消费堆积 > 10000 持续 5min
### 4.2 Alertmanager 路由
- P0/P1 → PagerDuty + 电话
- P2 → 飞书群 + 邮件
- P3 → 飞书群
- 抑制:同一服务 5min 内同告警只发一次inhibit + group_by
### 4.3 Grafana 仪表盘
| 仪表盘 | 用途 | 关键面板 |
|--------|------|----------|
| Service Overview | 服务总览 | QPS、P99、错误率、熔断状态 |
| SLO Dashboard | SLO 跟踪 | 可用性、错误预算消耗 |
| Dependency Health | 依赖健康 | DB/Redis/Kafka 连接与延迟 |
| Backup Status | 备份状态 | 最近备份、RPO、恢复演练 |
## 5. 混沌工程
### 5.1 实验清单
| 实验 | 注入方式 | 预期表现 | 频率 |
|------|----------|----------|------|
| DB 主节点宕机 | kill postgres | 自动故障转移RTO<30min | 月 |
| Redis 不可达 | iptables 拒绝 | 降级到本地缓存 | 月 |
| Kafka broker 宕机 | kill broker | 生产重试,消费堆积可控 | 月 |
| 下游服务延迟 | tc netem +500ms | 熔断器打开,错误率<1% | 周 |
| 网络分区 | iptables 隔离 | 多可用区切换 | 季 |
| 磁盘满 | fill disk | 告警触发,写入降级 | 季 |
### 5.2 执行流程
1. 选择实验 → 在 `chaos/` 选择对应 YAML
2. 预检:确认告警通道、回滚脚本就绪
3. 执行:`kubectl apply -f chaos/<experiment>.yaml`
4. 观察Grafana + 日志
5. 回滚:`./chaos/rollback.sh <experiment>`
6. 复盘:记录到 `docs/architecture/runbooks/incident-response.md`
## 6. 安全加固
### 6.1 WAF 规则
- SQL 注入、XSS 模式匹配
- 路径穿越、命令注入
- 限速:单 IP > 100req/s 拦截
- 地域封禁(按需)
### 6.2 密钥管理
- 密钥存储K8s Secret + 外部 KMSVault
- 轮换DB 密码每 90 天JWT 签名密钥每 180 天
- 注入:通过环境变量 / 挂载卷,禁止入镜像
- 审计:所有密钥访问记录到 audit log
### 6.3 CORS 策略
- 允许来源:白名单域名(生产环境严格)
- 允许方法GET/POST/PUT/PATCH/DELETE
- 凭证允许Cookie
- 预检缓存600s
## 7. K8s 部署
### 7.1 Manifest 说明
位置:`deploy/k8s/`,每个服务一组 manifest
- `deployment.yaml`:副本数、资源、探针、优雅停机
- `service.yaml`ClusterIP
- `hpa.yaml`CPU>70% 扩容min=3 max=20
- `poddisruptionbudget.yaml`minAvailable=2
- `networkpolicy.yaml`:限制出向
探针配置:
```yaml
livenessProbe:
httpGet: { path: /healthz, port: 3000 }
initialDelaySeconds: 15
periodSeconds: 10
readinessProbe:
httpGet: { path: /readyz, port: 3000 }
initialDelaySeconds: 5
periodSeconds: 5
```
优雅停机:
```yaml
terminationGracePeriodSeconds: 60
```
### 7.2 Helm 化路线
- P6原生 manifest快速验证
- P7抽取 Helm Chartvalues.yaml 按环境区分
- P8引入 Argo CD GitOps 自动同步
## 8. 灾难恢复
### 8.1 RTO/RPO 目标
| 场景 | RTO | RPO |
|------|-----|-----|
| 单 Pod 故障 | 30s | 0 |
| 单节点故障 | 2min | 0 |
| 单可用区故障 | 10min | 0 |
| 区域级灾难 | 30min | 15min |
### 8.2 多可用区策略
- K8s 集群跨 3 可用区Pod 反亲和
- DB 主从跨可用区同步复制
- Redis 哨兵跨可用区
- Kafka min.insync.replicas=2跨可用区 broker
### 8.3 DNS 切换
- 区域级故障:通过全局 DNSCloudflare/Route53切换到备用区域
- 健康检查:每 10s 探测,连续 3 次失败自动切换
- TTL60s快速切换
- 演练:每季度一次 DNS 切换演练
## 9. 故障排查
| 现象 | 可能原因 | 排查步骤 | 解决方案 |
|------|----------|----------|----------|
| 5xx 激增 | 下游服务故障 | 查 Grafana 熔断状态、下游健康 | 确认熔断器已打开,扩容下游 |
| P99 升高 | DB 慢查询/连接耗尽 | 查 PG 慢日志、连接池 | 加索引/扩连接池/限流 |
| 消费堆积 | 消费者慢/宕机 | 查 Kafka lag、消费者日志 | 扩消费者、修 bug |
| 备份失败 | 存储/网络/凭证 | 查 backup-cron 日志 | 修凭证、清理旧备份 |
| 健康检查失败 | DB 不可达 | 查 DB 状态、网络 | 故障转移、恢复 DB |
| Pod 频繁重启 | OOM/探针失败 | 查 kubectl describe、内存 | 调资源/修探针 |
| 熔断不恢复 | 下游未恢复 | 查 Half-Open 试探结果 | 修复下游、调 Timeout |
| 限流误杀 | 桶配置过低 | 查限流日志、QPS | 调高桶容量 |
| DNS 切换无效 | TTL 缓存 | 查 DNS 解析链 | 等待 TTL / 清缓存 |
| 跨区延迟高 | 跨区流量 | 查网络拓扑 | 调亲和性就近访问 |
---
## 附录:关联文档
- 架构影响地图:`docs/architecture/004_architecture_impact_map.md`
- P6 架构补记:`docs/architecture/004-p6-addendum.md`
- 事件响应:`docs/architecture/runbooks/incident-response.md`
- 已知问题:`docs/troubleshooting/known-issues.md`
- P6 已知问题补丁:`docs/troubleshooting/known-issues-p6-addendum.md`
- 技术债务:`docs/architecture/roadmap/tech-debt.md`

View File

@@ -0,0 +1,24 @@
# known-issues P6 补丁
> 本文件为 `docs/troubleshooting/known-issues.md` 的 P6 阶段补丁,列出 P6 新增的"场景→技术"映射。
> 合并方式:将下表条目追加到原文件对应分区,遵循索引式速查规范,不写代码示例。
## P6 生产硬化新增场景
| 场景 | 技术方案 |
|------|----------|
| 熔断器状态切换Closed/Open/Half-Open | gobreaker v2 ReadyToTrip 回调按连续失败数 + 失败率判定 |
| 限流桶按租户维度清理 | sync.Map + ticker 周期清理过期桶,避免内存泄漏 |
| 备份脚本定时调度 | K8s CronJob + backup-cron.sh15min 一次满足 RPO |
| PostgreSQL WAL 归档恢复到时间点 | pg_receivewal 归档 + PITR 恢复 |
| Redis 增量备份 | BGSAVE 触发 + RDB 文件上传对象存储 |
| Kafka 消费位点快照 | __consumer_offsets topic dump 到对象存储 |
| 熔断器半开态试探限流 | MaxRequests 限制并发试探,避免恢复期二次过载 |
| 健康探针分流 liveness/readiness | /healthz 仅进程存活,/readyz 检查依赖,避免滚动重启雪崩 |
| 优雅停机等待 in-flight 请求 | app.enableShutdownHooks + terminationGracePeriodSeconds=60 |
| Kafka producer 关闭前 flush | producer.disconnect() 内部 flush避免消息丢失 |
| 数据源销毁顺序 | 先 Kafka、再 Redis、最后 DB避免反向依赖阻塞 |
| 混沌实验回滚 | rollback.sh 按实验名清理 tc/iptables 规则 |
| DNS 切换消除缓存 | TTL=60s + 等待 + 客户端清缓存,避免切换无效 |
| 告警抑制去重 | Alertmanager inhibit + group_by + group_wait |
| Python 服务就绪检查延迟初始化 | readyz 返回 ok + TODO避免启动期依赖未就绪导致探针失败 |

View File

@@ -0,0 +1,69 @@
# Alertmanager 配置 - Edu 平台
# 全局配置
global:
resolve_timeout: 5m
# 告警模板(可扩展)
templates:
- /etc/alertmanager/templates/*.tmpl
# 路由树
route:
# 顶层默认接收器
receiver: webhook-default
# 按 alertname + service 分组
group_by: ['alertname', 'service']
# 首次告警等待时间(聚合相同组)
group_wait: 30s
# 同组新告警发送间隔
group_interval: 5m
# 重复告警发送间隔
repeat_interval: 4h
routes:
# critical 告警走 webhook
- matchers:
- severity="critical"
receiver: webhook-default
continue: false
# warning 告警走 webhook
- matchers:
- severity="warning"
receiver: webhook-default
continue: false
# 接收器列表
receivers:
# 默认 Webhook 接收器(指向内部告警路由服务)
- name: webhook-default
webhook_configs:
- url: http://alert-router:5001/alert
send_resolved: true
max_alerts: 0
# 邮件接收器(注释示例,留作扩展)
# 启用前需在 global.smtp_* 配置 SMTP 服务器
# - name: mail-ops
# email_configs:
# - to: ops-team@example.com
# from: alertmanager@example.com
# smarthost: smtp.example.com:587
# auth_username: alertmanager@example.com
# auth_password: <SMTP_PASSWORD>
# require_tls: true
# headers:
# Subject: '[Edu Alert] {{ .GroupLabels.alertname }}'
# 钉钉/企业微信接收器(注释示例,留作扩展)
# - name: dingtalk-ops
# webhook_configs:
# - url: http://dingtalk-webhook:8060/dingtalk/ops/send
# send_resolved: true
# 抑制规则critical 告警抑制同服务同名的 warning 告警
inhibit_rules:
- source_matchers:
- severity="critical"
target_matchers:
- severity="warning"
# 相同 alertname + service 才抑制
equal: ['alertname', 'service']

View File

@@ -0,0 +1,74 @@
#!/bin/bash
# MySQL 备份 cron 调度入口
# 调用 backup-mysql.sh 备份所有微服务数据库,单个失败不阻塞其他
# 用法: backup-cron.sh [--keep <days>]
set -euo pipefail
# ---------- 参数解析 ----------
KEEP_DAYS=7
while [[ $# -gt 0 ]]; do
case "$1" in
--keep)
KEEP_DAYS="${2:-7}"
shift 2
;;
-h|--help)
echo "Usage: $0 [--keep <days>]" >&2
echo " --keep 保留天数(默认 7" >&2
exit 0
;;
*)
echo "[ERROR] 未知参数: $1" >&2
exit 1
;;
esac
done
# ---------- 配置 ----------
# 每个服务对应一个独立数据库
SERVICES=(iam core-edu content msg classes)
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
BACKUP_SCRIPT="${SCRIPT_DIR}/backup-mysql.sh"
# ---------- 日志函数 ----------
log() {
echo "[$(date '+%Y-%m-%d %H:%M:%S')] $*" >&2
}
if [[ ! -x "$BACKUP_SCRIPT" ]]; then
log "[ERROR] 备份脚本不存在或不可执行: ${BACKUP_SCRIPT}"
exit 1
fi
# ---------- 主流程 ----------
log "[INFO] 开始批量备份,共 ${#SERVICES[@]} 个服务"
TOTAL=0
SUCCESS=0
FAILED=0
FAILED_LIST=()
for svc in "${SERVICES[@]}"; do
TOTAL=$((TOTAL + 1))
log "[INFO] ---- 备份服务: ${svc} ----"
if "$BACKUP_SCRIPT" --service "$svc" --keep "$KEEP_DAYS"; then
SUCCESS=$((SUCCESS + 1))
log "[INFO] 服务 ${svc} 备份成功"
else
FAILED=$((FAILED + 1))
FAILED_LIST+=("$svc")
log "[ERROR] 服务 ${svc} 备份失败,继续下一个"
fi
done
# ---------- 汇总 ----------
log "[INFO] 批量备份结束: 总计=${TOTAL} 成功=${SUCCESS} 失败=${FAILED}"
if [[ ${FAILED} -gt 0 ]]; then
log "[ERROR] 失败服务列表: ${FAILED_LIST[*]}"
exit 1
fi
exit 0

View File

@@ -0,0 +1,99 @@
#!/bin/bash
# MySQL 全量备份脚本Linux 容器内运行)
# 用法: backup-mysql.sh --service <name> [--keep <days>]
set -euo pipefail
# ---------- 参数解析 ----------
SERVICE=""
KEEP_DAYS=7
while [[ $# -gt 0 ]]; do
case "$1" in
--service)
SERVICE="${2:-}"
shift 2
;;
--keep)
KEEP_DAYS="${2:-7}"
shift 2
;;
-h|--help)
echo "Usage: $0 --service <name> [--keep <days>]" >&2
echo " --service 目标微服务数据库名(必填,如 iam/core-edu/content/msg/classes" >&2
echo " --keep 保留天数(默认 7" >&2
exit 0
;;
*)
echo "[ERROR] 未知参数: $1" >&2
exit 1
;;
esac
done
# ---------- 参数校验 ----------
if [[ -z "$SERVICE" ]]; then
echo "[ERROR] --service 参数必填" >&2
exit 1
fi
if ! [[ "$KEEP_DAYS" =~ ^[0-9]+$ ]] || [[ "$KEEP_DAYS" -lt 1 ]]; then
echo "[ERROR] --keep 必须为正整数" >&2
exit 1
fi
# ---------- 环境变量 ----------
: "${MYSQL_HOST:=mysql}"
: "${MYSQL_PORT:=3306}"
: "${MYSQL_USER:=root}"
: "${MYSQL_PASSWORD:?MYSQL_PASSWORD 环境变量未设置}"
BACKUP_ROOT="/backups"
BACKUP_DIR="${BACKUP_ROOT}/${SERVICE}"
TIMESTAMP="$(date +%Y%m%d-%H%M%S)"
BACKUP_FILE="${BACKUP_DIR}/${TIMESTAMP}.sql.gz"
# ---------- 日志函数 ----------
log() {
echo "[$(date '+%Y-%m-%d %H:%M:%S')] $*" >&2
}
# ---------- 主流程 ----------
log "[INFO] 开始备份服务: ${SERVICE} (保留 ${KEEP_DAYS} 天)"
mkdir -p "$BACKUP_DIR"
log "[INFO] 执行 mysqldump -> ${BACKUP_FILE}"
if ! mysqldump \
--host="$MYSQL_HOST" \
--port="$MYSQL_PORT" \
--user="$MYSQL_USER" \
--password="$MYSQL_PASSWORD" \
--single-transaction \
--routines \
--triggers \
--databases "$SERVICE" \
2>/dev/null \
| gzip -c > "$BACKUP_FILE"; then
log "[ERROR] mysqldump 失败 (service=${SERVICE})"
rm -f "$BACKUP_FILE"
exit 1
fi
# 校验产物非空
if [[ ! -s "$BACKUP_FILE" ]]; then
log "[ERROR] 备份文件为空: ${BACKUP_FILE}"
exit 1
fi
FILE_SIZE="$(stat -c %s "$BACKUP_FILE" 2>/dev/null || stat -f %z "$BACKUP_FILE")"
log "[INFO] 备份完成: ${BACKUP_FILE} (${FILE_SIZE} bytes)"
# ---------- 清理过期备份 ----------
log "[INFO] 清理超过 ${KEEP_DAYS} 天的旧备份"
find "$BACKUP_DIR" -type f -name '*.sql.gz' -mtime +${KEEP_DAYS} -print -delete \
| while read -r old_file; do
log "[INFO] 已删除: ${old_file}"
done
log "[INFO] 备份流程结束: ${SERVICE}"
exit 0

View File

@@ -0,0 +1,94 @@
#!/bin/bash
# MySQL 恢复脚本Linux 容器内运行)
# 用法: restore-mysql.sh --service <name> --file <path> [--yes]
set -euo pipefail
# ---------- 参数解析 ----------
SERVICE=""
BACKUP_FILE=""
ASSUME_YES=false
while [[ $# -gt 0 ]]; do
case "$1" in
--service)
SERVICE="${2:-}"
shift 2
;;
--file)
BACKUP_FILE="${2:-}"
shift 2
;;
--yes)
ASSUME_YES=true
shift
;;
-h|--help)
echo "Usage: $0 --service <name> --file <path> [--yes]" >&2
echo " --service 目标微服务数据库名(必填)" >&2
echo " --file 备份文件路径 sql.gz必填" >&2
echo " --yes 跳过确认提示" >&2
exit 0
;;
*)
echo "[ERROR] 未知参数: $1" >&2
exit 1
;;
esac
done
# ---------- 参数校验 ----------
if [[ -z "$SERVICE" ]]; then
echo "[ERROR] --service 参数必填" >&2
exit 1
fi
if [[ -z "$BACKUP_FILE" ]]; then
echo "[ERROR] --file 参数必填" >&2
exit 1
fi
if [[ ! -f "$BACKUP_FILE" ]]; then
echo "[ERROR] 备份文件不存在: ${BACKUP_FILE}" >&2
exit 1
fi
# ---------- 环境变量 ----------
: "${MYSQL_HOST:=mysql}"
: "${MYSQL_PORT:=3306}"
: "${MYSQL_USER:=root}"
: "${MYSQL_PASSWORD:?MYSQL_PASSWORD 环境变量未设置}"
# ---------- 日志函数 ----------
log() {
echo "[$(date '+%Y-%m-%d %H:%M:%S')] $*" >&2
}
# ---------- 确认提示 ----------
log "[WARN] 即将向数据库 [${SERVICE}] 导入备份文件: ${BACKUP_FILE}"
log "[WARN] 此操作会覆盖目标数据库现有数据,不可撤销!"
if [[ "$ASSUME_YES" != "true" ]]; then
read -r -p "确认继续? (输入 YES 继续): " confirm
if [[ "$confirm" != "YES" ]]; then
log "[INFO] 用户取消操作"
exit 0
fi
fi
# ---------- 主流程 ----------
log "[INFO] 开始恢复服务: ${SERVICE}"
log "[INFO] 解压并导入: ${BACKUP_FILE}"
if ! gunzip -c "$BACKUP_FILE" \
| mysql \
--host="$MYSQL_HOST" \
--port="$MYSQL_PORT" \
--user="$MYSQL_USER" \
--password="$MYSQL_PASSWORD" \
"$SERVICE"; then
log "[ERROR] mysql 导入失败 (service=${SERVICE})"
exit 1
fi
log "[INFO] 恢复完成: ${SERVICE} <- ${BACKUP_FILE}"
exit 0

45
infra/chaos/README.md Normal file
View File

@@ -0,0 +1,45 @@
# 混沌工程说明
## 适用范围
- **仅 Staging 环境**:禁止在生产环境运行混沌实验
- **每月 1 次演练**:建议每月第一个工作周执行
- **演练窗口**业务低峰期02:00 - 05:00
## 实验清单
| 实验 | 注入故障 | 验证目标 |
|------|----------|----------|
| `pod-delete` | 杀死 50% Pod | 自愈能力、副本数冗余 |
| `pod-network-latency` | 注入 200ms 延迟 | 超时与重试、降级策略 |
| `disk-fill` | 填充磁盘至 80% | 日志写入、磁盘告警 |
## 前置条件
1. Staging 环境已部署完整监控Prometheus + Alertmanager + Grafana
2. 已配置稳态探针(`steadyStateHypothesis`
3. 已配置告警通道并验证可达
4. 演练参与者已就位(运维 + 研发 oncall
## 演练流程
1. **演练前 30 分钟**:通知相关人员,确认稳态基线
2. **启动实验**:按顺序执行,单实验单次注入
3. **观察监控**:关注告警是否按预期触发、服务是否自愈
4. **演练后**:归档实验结果,更新本目录 README
## 回滚预案
- **5 分钟内未恢复自动回滚**:通过 Litmus `steadyStateHypothesis` 失败触发实验终止
- 手动回滚:
```bash
kubectl delete chaosengine edu-chaos-experiments -n edu-monitoring
kubectl rollout restart deployment/api-gateway -n edu-services
```
- 极端情况:直接缩容 / 扩容受影响 Deployment
## 安全约束
- 每次实验**只注入一种故障**,避免叠加影响判断
- 实验前快照数据库(参考 `infra/backup/backup-cron.sh`
- 实验期间禁止发布任何业务变更

View File

@@ -0,0 +1,139 @@
# 混沌实验配置Litmus Chaos 骨架)
# 仅作 YAML 骨架,注释说明用途。实际运行需配合 Litmus Chaos Operator。
# 文档https://litmuschaos.github.io/litmus/
apiVersion: litmuschaos.io/v1alpha1
kind: ChaosEngine
metadata:
name: edu-chaos-experiments
namespace: edu-monitoring
labels:
app.kubernetes.io/part-of: edu-platform
app.kubernetes.io/component: chaos
spec:
appinfo:
appns: edu-services
applabel: "app.kubernetes.io/name=api-gateway"
appkind: deployment
chaosServiceAccount: litmus-chaos-sa
components:
experiments:
# ============================================================
# 实验 1服务 Pod 杀死(验证自愈与副本数)
# ============================================================
- name: pod-delete
spec:
components:
env:
# 杀死 1 个 Pod
- name: TOTAL_CHAOS_DURATION
value: "30"
- name: CHAOS_INTERVAL
value: "10"
- name: FORCE
value: "false"
- name: PODS_AFFECTED_PERC
value: "50"
- name: TARGET_CONTAINER
value: "api-gateway"
# 假设:杀死 50% Pod 后,服务仍可对外可用(最少 1 个副本健康)
# 稳态up{job="api-gateway"} >= 1
steadyStateHypothesis:
steadyStateHypothesis:
probe:
- name: api-gateway-still-up
type: httpProbe
mode: Continuous
runProperties:
probeTimeout: 5
httpProbeInputs:
url: http://api-gateway.edu-services.svc:8080/healthz
method:
get: {}
phases:
- name: pre-chaos
description: "混沌前稳态验证"
- name: inject
description: "注入 Pod 删除"
- name: post-chaos
description: "混沌后自愈验证"
# ============================================================
# 实验 2网络延迟注入验证超时与重试
# ============================================================
- name: pod-network-latency
spec:
components:
env:
# 注入 200ms 网络延迟
- name: NETWORK_LATENCY
value: "200"
- name: TOTAL_CHAOS_DURATION
value: "60"
- name: CHAOS_INTERVAL
value: "10"
- name: NETWORK_INTERFACE
value: "eth0"
- name: TARGET_CONTAINER
value: "api-gateway"
# 假设200ms 延迟下 P99 < 2s无 5xx 雪崩
# 稳态:错误率 < 5%
steadyStateHypothesis:
steadyStateHypothesis:
probe:
- name: error-rate-below-threshold
type: promProbe
mode: Continuous
runProperties:
probeTimeout: 5
promProbeInputs:
source:
url: http://prometheus.edu-monitoring.svc:9090
query: |
sum(rate(http_requests_total{service="api-gateway",status=~"5.."}[1m]))
/ sum(rate(http_requests_total{service="api-gateway"}[1m]))
comparator:
criteria: "<="
value: "0.05"
phases:
- name: pre-chaos
description: "混沌前稳态验证"
- name: inject
description: "注入 200ms 网络延迟"
- name: post-chaos
description: "混沌后恢复验证"
# ============================================================
# 实验 3磁盘填充验证磁盘压力下日志写入与告警
# ============================================================
- name: disk-fill
spec:
components:
env:
# 填充至 80% 磁盘使用率
- name: FILL_PERCENTAGE
value: "80"
- name: TOTAL_CHAOS_DURATION
value: "120"
- name: CHAOS_INTERVAL
value: "30"
- name: TARGET_CONTAINER
value: "api-gateway"
# 假设:磁盘 80% 使用率下服务仍可写入日志,触发 DiskSpaceLow 告警
# 稳态:服务 /healthz 可用
steadyStateHypothesis:
steadyStateHypothesis:
probe:
- name: api-gateway-healthz
type: httpProbe
mode: Continuous
runProperties:
probeTimeout: 5
httpProbeInputs:
url: http://api-gateway.edu-services.svc:8080/healthz
method:
get: {}
phases:
- name: pre-chaos
description: "混沌前稳态验证"
- name: inject
description: "填充磁盘至 80%"
- name: post-chaos
description: "混沌后清理与恢复验证"

View File

@@ -0,0 +1,119 @@
# 监控栈 docker-compose独立 profile
# 使用方式: docker compose -f docker-compose.monitoring.yml --profile monitoring up -d
version: "3.9"
networks:
edu-network:
name: edu-network
external: true
volumes:
prometheus-data:
name: edu-prometheus-data
alertmanager-data:
name: edu-alertmanager-data
grafana-data:
name: edu-grafana-data
services:
# ============================================================
# Prometheus - 指标采集与存储
# ============================================================
prometheus:
image: prom/prometheus:v2.54.1
container_name: edu-prometheus
profiles: ["monitoring"]
restart: unless-stopped
command:
- "--config.file=/etc/prometheus/prometheus.yml"
- "--storage.tsdb.path=/prometheus"
- "--storage.tsdb.retention.time=15d"
- "--web.enable-lifecycle"
- "--web.enable-admin-api"
ports:
- "9090:9090"
volumes:
- ./prometheus.yml:/etc/prometheus/prometheus.yml:ro
- ./prometheus/rules.yml:/etc/prometheus/rules.yml:ro
- prometheus-data:/prometheus
networks:
- edu-network
# ============================================================
# Alertmanager - 告警路由与抑制
# ============================================================
alertmanager:
image: prom/alertmanager:v0.27.0
container_name: edu-alertmanager
profiles: ["monitoring"]
restart: unless-stopped
command:
- "--config.file=/etc/alertmanager/alertmanager.yml"
- "--storage.path=/alertmanager"
ports:
- "9093:9093"
volumes:
- ./alertmanager/alertmanager.yml:/etc/alertmanager/alertmanager.yml:ro
- alertmanager-data:/alertmanager
networks:
- edu-network
depends_on:
- prometheus
# ============================================================
# Grafana - 可视化
# ============================================================
grafana:
image: grafana/grafana:11.2.2
container_name: edu-grafana
profiles: ["monitoring"]
restart: unless-stopped
environment:
- GF_SECURITY_ADMIN_USER=${GRAFANA_ADMIN_USER:-admin}
- GF_SECURITY_ADMIN_PASSWORD=${GRAFANA_ADMIN_PASSWORD:-admin}
- GF_USERS_ALLOW_SIGN_UP=false
- GF_AUTH_ANONYMOUS_ENABLED=false
ports:
- "3000:3000"
volumes:
- ./grafana/provisioning:/etc/grafana/provisioning:ro
- ./grafana/dashboards:/var/lib/grafana/dashboards:ro
- grafana-data:/var/lib/grafana
networks:
- edu-network
depends_on:
- prometheus
# ============================================================
# node-exporter - 主机指标采集
# ============================================================
node-exporter:
image: prom/node-exporter:v1.8.2
container_name: edu-node-exporter
profiles: ["monitoring"]
restart: unless-stopped
command:
- "--path.rootfs=/host"
ports:
- "9100:9100"
volumes:
- /proc:/host/proc:ro
- /sys:/host/sys:ro
- /:/host:ro
networks:
- edu-network
# ============================================================
# blackbox-exporter - 黑盒探测HTTP / TCP / ICMP
# ============================================================
blackbox-exporter:
image: prom/blackbox-exporter:v0.25.0
container_name: edu-blackbox-exporter
profiles: ["monitoring"]
restart: unless-stopped
ports:
- "9115:9115"
volumes:
- ./blackbox/blackbox.yml:/etc/blackbox_exporter/config.yml:ro
networks:
- edu-network

View File

@@ -0,0 +1,394 @@
{
"annotations": {
"list": [
{
"builtIn": 1,
"datasource": {
"type": "grafana",
"uid": "-- Grafana --"
},
"enable": true,
"hide": true,
"iconColor": "rgba(0, 211, 255, 1)",
"name": "Annotations & Alerts",
"type": "dashboard"
}
]
},
"description": "Edu 平台微服务总览 - QPS / 错误率 / P99 延迟 / 熔断器状态",
"editable": true,
"fiscalYearStartMonth": 0,
"graphTooltip": 1,
"id": null,
"links": [],
"liveNow": false,
"panels": [
{
"datasource": {
"type": "prometheus",
"uid": "prometheus"
},
"fieldConfig": {
"defaults": {
"color": {
"mode": "palette-classic"
},
"custom": {
"axisCenteredZero": false,
"axisColorMode": "text",
"axisLabel": "",
"axisPlacement": "auto",
"barAlignment": 0,
"drawStyle": "line",
"fillOpacity": 10,
"gradientMode": "none",
"hideFrom": {
"legend": false,
"tooltip": false,
"viz": false
},
"insertNulls": false,
"lineInterpolation": "linear",
"lineWidth": 2,
"pointSize": 5,
"scaleDistribution": {
"type": "linear"
},
"showPoints": "never",
"spanNulls": true,
"stacking": {
"group": "A",
"mode": "none"
},
"thresholdsStyle": {
"mode": "off"
}
},
"mappings": [],
"thresholds": {
"mode": "absolute",
"steps": [
{
"color": "green",
"value": null
}
]
},
"unit": "reqps"
},
"overrides": []
},
"gridPos": {
"h": 8,
"w": 12,
"x": 0,
"y": 0
},
"id": 1,
"options": {
"legend": {
"calcs": ["mean", "max"],
"displayMode": "table",
"placement": "bottom",
"showLegend": true
},
"tooltip": {
"mode": "multi",
"sort": "desc"
}
},
"targets": [
{
"datasource": {
"type": "prometheus",
"uid": "prometheus"
},
"expr": "sum(rate(http_requests_total[1m])) by (service)",
"legendFormat": "{{service}}",
"refId": "A"
}
],
"title": "请求 QPS按服务",
"type": "timeseries"
},
{
"datasource": {
"type": "prometheus",
"uid": "prometheus"
},
"fieldConfig": {
"defaults": {
"color": {
"mode": "palette-classic"
},
"custom": {
"axisCenteredZero": false,
"axisColorMode": "text",
"axisLabel": "",
"axisPlacement": "auto",
"barAlignment": 0,
"drawStyle": "line",
"fillOpacity": 20,
"gradientMode": "none",
"hideFrom": {
"legend": false,
"tooltip": false,
"viz": false
},
"insertNulls": false,
"lineInterpolation": "linear",
"lineWidth": 2,
"pointSize": 5,
"scaleDistribution": {
"type": "linear"
},
"showPoints": "never",
"spanNulls": true,
"stacking": {
"group": "A",
"mode": "none"
},
"thresholdsStyle": {
"mode": "line"
}
},
"mappings": [],
"thresholds": {
"mode": "absolute",
"steps": [
{
"color": "green",
"value": null
},
{
"color": "red",
"value": 0.05
}
]
},
"unit": "percentunit"
},
"overrides": []
},
"gridPos": {
"h": 8,
"w": 12,
"x": 12,
"y": 0
},
"id": 2,
"options": {
"legend": {
"calcs": ["mean", "max"],
"displayMode": "table",
"placement": "bottom",
"showLegend": true
},
"tooltip": {
"mode": "multi",
"sort": "desc"
}
},
"targets": [
{
"datasource": {
"type": "prometheus",
"uid": "prometheus"
},
"expr": "sum(rate(http_requests_total{status=~\"5..\"}[5m])) by (service) / sum(rate(http_requests_total[5m])) by (service)",
"legendFormat": "{{service}} 5xx",
"refId": "A"
}
],
"title": "错误率5xx 占比)",
"type": "timeseries"
},
{
"datasource": {
"type": "prometheus",
"uid": "prometheus"
},
"fieldConfig": {
"defaults": {
"color": {
"mode": "palette-classic"
},
"custom": {
"axisCenteredZero": false,
"axisColorMode": "text",
"axisLabel": "",
"axisPlacement": "auto",
"barAlignment": 0,
"drawStyle": "line",
"fillOpacity": 10,
"gradientMode": "none",
"hideFrom": {
"legend": false,
"tooltip": false,
"viz": false
},
"insertNulls": false,
"lineInterpolation": "linear",
"lineWidth": 2,
"pointSize": 5,
"scaleDistribution": {
"type": "linear"
},
"showPoints": "never",
"spanNulls": true,
"stacking": {
"group": "A",
"mode": "none"
},
"thresholdsStyle": {
"mode": "line"
}
},
"mappings": [],
"thresholds": {
"mode": "absolute",
"steps": [
{
"color": "green",
"value": null
},
{
"color": "orange",
"value": 1
}
]
},
"unit": "s"
},
"overrides": []
},
"gridPos": {
"h": 8,
"w": 12,
"x": 0,
"y": 8
},
"id": 3,
"options": {
"legend": {
"calcs": ["mean", "max"],
"displayMode": "table",
"placement": "bottom",
"showLegend": true
},
"tooltip": {
"mode": "multi",
"sort": "desc"
}
},
"targets": [
{
"datasource": {
"type": "prometheus",
"uid": "prometheus"
},
"expr": "histogram_quantile(0.99, sum(rate(http_request_duration_seconds_bucket[5m])) by (le, service))",
"legendFormat": "{{service}} P99",
"refId": "A"
}
],
"title": "P99 延迟",
"type": "timeseries"
},
{
"datasource": {
"type": "prometheus",
"uid": "prometheus"
},
"fieldConfig": {
"defaults": {
"color": {
"mode": "thresholds"
},
"mappings": [
{
"options": {
"0": {
"color": "green",
"index": 0,
"text": "CLOSED"
},
"1": {
"color": "red",
"index": 1,
"text": "OPEN"
}
},
"type": "value"
}
],
"thresholds": {
"mode": "absolute",
"steps": [
{
"color": "green",
"value": null
},
{
"color": "red",
"value": 1
}
]
},
"unit": "none"
},
"overrides": []
},
"gridPos": {
"h": 8,
"w": 12,
"x": 12,
"y": 8
},
"id": 4,
"options": {
"colorMode": "value",
"graphMode": "none",
"justifyMode": "auto",
"orientation": "horizontal",
"reduceOptions": {
"calcs": ["lastNotNull"],
"fields": "",
"values": false
},
"showPercentChange": false,
"textMode": "auto"
},
"targets": [
{
"datasource": {
"type": "prometheus",
"uid": "prometheus"
},
"expr": "max(circuit_breaker_state{state=\"open\"}) by (service)",
"legendFormat": "{{service}}",
"refId": "A"
}
],
"title": "熔断器状态0=CLOSED, 1=OPEN",
"type": "stat"
}
],
"refresh": "30s",
"schemaVersion": 39,
"style": "dark",
"tags": ["edu", "microservices", "overview"],
"templating": {
"list": []
},
"time": {
"from": "now-1h",
"to": "now"
},
"timepicker": {},
"timezone": "",
"title": "Edu 微服务总览",
"uid": "edu-microservices-overview",
"version": 1,
"weekStart": ""
}

View File

@@ -0,0 +1,15 @@
# Grafana 仪表盘 provisioning
apiVersion: 1
providers:
- name: edu-dashboards
orgId: 1
folder: "Edu"
type: file
disableDeletion: false
updateIntervalSeconds: 30
allowUiUpdates: true
options:
# 从该目录加载所有仪表盘 JSON
path: /var/lib/grafana/dashboards
foldersFromFilesStructure: false

View File

@@ -0,0 +1,15 @@
# Grafana 数据源 provisioning
apiVersion: 1
datasources:
- name: Prometheus
type: prometheus
uid: prometheus
access: proxy
url: http://prometheus:9090
isDefault: true
editable: false
jsonData:
timeInterval: "15s"
httpMethod: POST
manageAlerts: false

55
infra/k8s/README.md Normal file
View File

@@ -0,0 +1,55 @@
# K8s 部署说明(骨架)
> **说明**:本目录为 K8s 部署**骨架**,仅提供最小可用 manifest**生产环境请使用 Helm Chart**。
## 适用范围
- 开发 / Staging 环境快速拉起
- 验证服务在 K8s 上的基本可用性
- 作为后续 Helm 化的过渡参考
## Manifest 类别
| 文件 | 类别 | 用途 |
|------|------|------|
| `namespace.yaml` | namespace | 划分 4 个命名空间system / services / monitoring / ingress |
| `configmap-*.yaml` | configmap | 非敏感配置(服务端口、特性开关等) |
| `secret-*.yaml` | secret | 敏感配置(密钥、连接串),生产用 External Secrets |
| `*-deployment.yaml` | deployment | 服务无状态部署,含探针 / 资源 / 副本数 |
| `*-service.yaml` | service | ClusterIP 服务发现 |
| `ingress.yaml` | ingress | 对外入口TLS 终结 + 路由 |
| `*-hpa.yaml` | hpa | 水平自动扩缩容 |
## 命名空间规划
| Namespace | 用途 |
|-----------|------|
| `edu-system` | 系统组件(数据库代理、配置等) |
| `edu-services` | 业务微服务api-gateway / iam / core-edu / content / msg |
| `edu-monitoring` | 监控栈Prometheus / Grafana / Alertmanager |
| `edu-ingress` | 入口控制器NGINX Ingress / cert-manager |
## 部署顺序
```bash
kubectl apply -f namespace.yaml
kubectl apply -f configmap-*.yaml
kubectl apply -f secret-*.yaml # 生产请改用 External Secrets
kubectl apply -f *-deployment.yaml
kubectl apply -f *-service.yaml
kubectl apply -f ingress.yaml
kubectl apply -f *-hpa.yaml
```
## Helm 化路线(后续)
1. 将骨架迁移为 Helm Chart`charts/edu-platform/`
2. 每个微服务一个子 Chart统一通过 umbrella chart 编排
3. 环境差异通过 values-<env>.yaml 管理
4. 敏感配置接入 External Secrets Operator对接 Vault / KMS
5. CI/CD 通过 ArgoCD / Flux 做 GitOps 部署
## 当前文件
- `namespace.yaml`4 个命名空间定义
- `api-gateway-deployment.yaml`api-gateway Deployment + Service 骨架

View File

@@ -0,0 +1,125 @@
# api-gateway Deployment + Service 骨架
# 生产环境请通过 Helm Chart 管理,此处仅作骨架参考
apiVersion: apps/v1
kind: Deployment
metadata:
name: api-gateway
namespace: edu-services
labels:
app.kubernetes.io/name: api-gateway
app.kubernetes.io/part-of: edu-platform
app.kubernetes.io/component: gateway
spec:
replicas: 2
selector:
matchLabels:
app.kubernetes.io/name: api-gateway
strategy:
type: RollingUpdate
rollingUpdate:
maxSurge: 1
maxUnavailable: 0
template:
metadata:
labels:
app.kubernetes.io/name: api-gateway
app.kubernetes.io/part-of: edu-platform
app.kubernetes.io/component: gateway
annotations:
prometheus.io/scrape: "true"
prometheus.io/port: "8080"
prometheus.io/path: "/metrics"
spec:
containers:
- name: api-gateway
image: edu/api-gateway:latest # 生产请固定 tag避免 latest
imagePullPolicy: IfNotPresent
ports:
- name: http
containerPort: 8080
protocol: TCP
# 存活探针:失败触发重启
livenessProbe:
httpGet:
path: /healthz
port: http
initialDelaySeconds: 15
periodSeconds: 20
timeoutSeconds: 3
failureThreshold: 3
# 就绪探针:失败从 Service Endpoints 摘除
readinessProbe:
httpGet:
path: /readyz
port: http
initialDelaySeconds: 5
periodSeconds: 10
timeoutSeconds: 3
failureThreshold: 2
resources:
requests:
cpu: "250m"
memory: "256Mi"
limits:
cpu: "1000m"
memory: "1Gi"
env:
# 从 ConfigMap 引用非敏感配置
- name: NODE_ENV
valueFrom:
configMapKeyRef:
name: api-gateway-config
key: NODE_ENV
- name: LOG_LEVEL
valueFrom:
configMapKeyRef:
name: api-gateway-config
key: LOG_LEVEL
- name: MYSQL_HOST
valueFrom:
configMapKeyRef:
name: api-gateway-config
key: MYSQL_HOST
# 从 Secret 引用敏感配置
- name: MYSQL_PASSWORD
valueFrom:
secretKeyRef:
name: api-gateway-secret
key: MYSQL_PASSWORD
- name: JWT_SECRET
valueFrom:
secretKeyRef:
name: api-gateway-secret
key: JWT_SECRET
- name: REDIS_PASSWORD
valueFrom:
secretKeyRef:
name: api-gateway-secret
key: REDIS_PASSWORD
# 生产建议挂载 /tmp 并设置 readOnlyRootFilesystem: true
securityContext:
runAsNonRoot: true
runAsUser: 1000
allowPrivilegeEscalation: false
capabilities:
drop: ["ALL"]
---
# api-gateway ServiceClusterIP
apiVersion: v1
kind: Service
metadata:
name: api-gateway
namespace: edu-services
labels:
app.kubernetes.io/name: api-gateway
app.kubernetes.io/part-of: edu-platform
app.kubernetes.io/component: gateway
spec:
type: ClusterIP
selector:
app.kubernetes.io/name: api-gateway
ports:
- name: http
port: 8080
targetPort: http
protocol: TCP

33
infra/k8s/namespace.yaml Normal file
View File

@@ -0,0 +1,33 @@
# K8s 命名空间定义 - Edu 平台
# 划分 4 个命名空间system / services / monitoring / ingress
apiVersion: v1
kind: Namespace
metadata:
name: edu-system
labels:
app.kubernetes.io/part-of: edu-platform
app.kubernetes.io/component: system
---
apiVersion: v1
kind: Namespace
metadata:
name: edu-services
labels:
app.kubernetes.io/part-of: edu-platform
app.kubernetes.io/component: services
---
apiVersion: v1
kind: Namespace
metadata:
name: edu-monitoring
labels:
app.kubernetes.io/part-of: edu-platform
app.kubernetes.io/component: monitoring
---
apiVersion: v1
kind: Namespace
metadata:
name: edu-ingress
labels:
app.kubernetes.io/part-of: edu-platform
app.kubernetes.io/component: ingress

122
infra/prometheus/rules.yml Normal file
View File

@@ -0,0 +1,122 @@
# Prometheus 告警规则 - Edu 平台
# 分组:服务可用性 / 性能 / 资源
groups:
# ============================================================
# 服务可用性
# ============================================================
- name: service-availability
rules:
# 服务宕机
- alert: ServiceDown
expr: up == 0
for: 1m
labels:
severity: critical
category: availability
annotations:
summary: "服务宕机 {{ $labels.instance }}"
description: "目标 {{ $labels.job }} / {{ $labels.instance }} 已离线超过 1 分钟。"
# 熔断器打开
- alert: CircuitBreakerOpen
expr: circuit_breaker_state{state="open"} == 1
for: 30s
labels:
severity: critical
category: availability
annotations:
summary: "熔断器打开 {{ $labels.service }}"
description: "服务 {{ $labels.service }} 熔断器处于 open 状态,可能正在拒绝请求。"
# ============================================================
# 性能
# ============================================================
- name: performance
rules:
# 高错误率5xx 占比 > 5%
- alert: HighErrorRate
expr: |
(sum(rate(http_requests_total{status=~"5.."}[5m])) by (service)
/ sum(rate(http_requests_total[5m])) by (service))
> 0.05
for: 5m
labels:
severity: warning
category: performance
annotations:
summary: "错误率过高 {{ $labels.service }}"
description: "服务 {{ $labels.service }} 5xx 错误率超过 5%,持续 5 分钟。"
# P99 延迟过高
- alert: HighLatencyP99
expr: |
histogram_quantile(0.99,
sum(rate(http_request_duration_seconds_bucket[5m])) by (le, service))
> 1
for: 5m
labels:
severity: warning
category: performance
annotations:
summary: "P99 延迟过高 {{ $labels.service }}"
description: "服务 {{ $labels.service }} P99 延迟超过 1 秒,持续 5 分钟。"
# Kafka 消费者积压
- alert: KafkaConsumerLag
expr: kafka_consumergroup_lag > 1000
for: 10m
labels:
severity: warning
category: performance
annotations:
summary: "Kafka 消费者积压 {{ $labels.consumergroup }}"
description: "消费组 {{ $labels.consumergroup }} 在 topic {{ $labels.topic }} 上积压超过 1000 条,持续 10 分钟。"
# ============================================================
# 资源
# ============================================================
- name: resources
rules:
# MySQL 连接数过高
- alert: MySQLConnectionsHigh
expr: |
(mysql_global_status_threads_connected
/ mysql_global_variables_max_connections)
> 0.8
for: 5m
labels:
severity: warning
category: resource
annotations:
summary: "MySQL 连接数过高 {{ $labels.instance }}"
description: "MySQL 实例 {{ $labels.instance }} 连接数超过最大连接数的 80%,持续 5 分钟。"
# 磁盘空间不足
- alert: DiskSpaceLow
expr: |
(100
- (node_filesystem_avail_bytes
/ node_filesystem_size_bytes * 100))
> 85
for: 10m
labels:
severity: warning
category: resource
annotations:
summary: "磁盘空间不足 {{ $labels.instance }}"
description: "节点 {{ $labels.instance }} 挂载点 {{ $labels.mountpoint }} 磁盘使用率超过 85%,持续 10 分钟。"
# 内存使用过高
- alert: MemoryHigh
expr: |
(1
- (node_memory_MemAvailable_bytes
/ node_memory_MemTotal_bytes))
> 0.9
for: 5m
labels:
severity: critical
category: resource
annotations:
summary: "内存使用过高 {{ $labels.instance }}"
description: "节点 {{ $labels.instance }} 内存使用率超过 90%,持续 5 分钟。"

View File

@@ -0,0 +1,50 @@
# Edu 平台密钥示例环境变量文件
# ============================================================
# 警告:本文件仅作示例,禁止包含真实密钥。
# 生产环境请通过 K8s Secret / External Secrets / Vault 注入。
# 复制为 .env 后用真实值替换所有 <placeholder>。
# ---------- MySQL ----------
# 用途MySQL root 用户密码,用于初始化与备份/恢复
# 最小长度32 字符
# 复杂度:含大小写字母 + 数字 + 特殊符号
MYSQL_ROOT_PASSWORD=<replace-with-32-char-strong-password>
# ---------- JWT ----------
# 用途JWT Access Token 签名密钥HS256
# 最小长度64 字符(建议使用 openssl rand -base64 48 生成)
# 注意:旋转后所有已签发的 Access Token 立即失效
JWT_SECRET=<replace-with-64-char-jwt-signing-secret>
# 用途JWT Refresh Token 签名密钥HS256
# 最小长度64 字符
# 注意:与 JWT_SECRET 必须不同;旋转后所有用户需重新登录
JWT_REFRESH_SECRET=<replace-with-64-char-refresh-signing-secret>
# ---------- Kafka ----------
# 用途Kafka SASL/PLAIN 认证密码
# 最小长度24 字符
KAFKA_SASL_PASSWORD=<replace-with-24-char-kafka-password>
# ---------- Elasticsearch ----------
# 用途Elasticsearch 内置 elastic 用户密码
# 最小长度24 字符
ES_PASSWORD=<replace-with-24-char-es-password>
# ---------- Neo4j ----------
# 用途Neo4j 数据库管理员密码
# 最小长度24 字符
NEO4J_PASSWORD=<replace-with-24-char-neo4j-password>
# ---------- Redis ----------
# 用途Redis ACL 默认用户密码
# 最小长度24 字符
# 注意:生产环境建议启用 ACL按用户分配最小权限
REDIS_PASSWORD=<replace-with-24-char-redis-password>
# ---------- 应用层加密 ----------
# 用途应用层字段级加密密钥AES-256-GCM
# 最小长度32 字节base64 编码后约 44 字符)
# 生成openssl rand -base64 32
# 注意:旋转前需先解密所有已加密字段,旋转后重新加密
ENCRYPTION_KEY=<replace-with-base64-32-byte-aes-key>

View File

@@ -0,0 +1,64 @@
# ModSecurity 兼容 WAF 规则骨架 - Edu 平台
# ============================================================
# 说明:本文件为 ModSecurity SecRule 语法骨架,
# 生产部署前请配合 OWASP CRS 使用并完成规则调优。
# 参考https://github.com/SpiderLabs/ModSecurity
# https://github.com/coreruleset/coreruleset
# ---------- 基础配置 ----------
# 启用规则引擎
SecRuleEngine On
# 请求体大小限制10MB
SecRequestBodyLimit 10485760
# 默认动作:记录日志 + 拒绝(生产前调整)
SecDefaultAction "phase:2,log,deny,status:403"
# ============================================================
# 规则 1SQL 注入检测
# 检测常见 SQL 注入关键字与元字符
# ============================================================
SecRule ARGS|ARGS_NAMES|REQUEST_COOKIES|REQUEST_COOKIES_NAMES|REQUEST_BODY|XML:/* \
"(?i)(union\s+select|select\s+.*\s+from|insert\s+into|update\s+.*\s+set|delete\s+from|drop\s+table|alter\s+table|create\s+table|exec\s*\(|;\s*drop\s|--\s|/\*.*\*/|or\s+1\s*=\s*1|and\s+1\s*=\s*1|'\s*or\s*'|sleep\s*\(|benchmark\s*\(|load_file\s*\(|into\s+outfile)" \
"id:1001,phase:2,log,deny,status:403,msg:'SQL Injection attempt',tag:'attack-sqli',severity:CRITICAL"
# ============================================================
# 规则 2XSS 检测
# 检测常见 XSS payload 与事件处理器
# ============================================================
SecRule ARGS|ARGS_NAMES|REQUEST_COOKIES|REQUEST_COOKIES_NAMES|REQUEST_BODY|XML:/* \
"(?i)(<script|</script>|javascript:|onerror\s*=|onload\s*=|onclick\s*=|onmouseover\s*=|onfocus\s*=|onblur\s*=|<iframe|<img[^>]+src\s*=|<svg[^>]+onload|document\.cookie|window\.location|eval\s*\(|alert\s*\(|prompt\s*\(|String\.fromCharCode)" \
"id:1002,phase:2,log,deny,status:403,msg:'XSS attempt',tag:'attack-xss',severity:CRITICAL"
# ============================================================
# 规则 3路径遍历检测
# 检测 ../ 与编码变体,防止读取敏感文件
# ============================================================
SecRule ARGS|ARGS_NAMES|REQUEST_FILENAME|REQUEST_URI|REQUEST_BODY \
"(?i)(\.\./|\.\.\\|\.\.%2f|\.\.%5c|%2e%2e%2f|%2e%2e%5c|/etc/passwd|/etc/shadow|/etc/hosts|/proc/self/environ|c:\\windows\\win\.ini|c:\\boot\.ini|\.\./\.\./\.\./)" \
"id:1003,phase:2,log,deny,status:403,msg:'Path Traversal attempt',tag:'attack-lfi',severity:CRITICAL"
# ============================================================
# 规则 4User-Agent 黑名单
# 拦截已知恶意扫描器与爬虫
# ============================================================
SecRule REQUEST_HEADERS:User-Agent \
"(?i)(sqlmap|nikto|nmap|nessus|acunetix|wpscan|hydra|metasploit|burpcollaborator|masscan|zgrab|dirbuster|gobuster|fuzzer|webinspect|appscan|jaeles|xrkmd)" \
"id:1004,phase:1,log,deny,status:403,msg:'Malicious User-Agent blocked',tag:'attack-scanner',severity:WARNING"
# ============================================================
# 规则 5远程文件包含RFI检测
# 检测通过 URL 参数引入远程文件
# ============================================================
SecRule ARGS|ARGS_NAMES \
"(?i)(^(http|https|ftp|php|data)://|(http|https|ftp|php|data)://.*\?(php|http|https|ftp|data)://)" \
"id:1005,phase:2,log,deny,status:403,msg:'Remote File Inclusion attempt',tag:'attack-rfi',severity:CRITICAL"
# ============================================================
# 规则 6命令注入检测
# 检测 shell 元字符与命令执行关键字
# ============================================================
SecRule ARGS|ARGS_NAMES|REQUEST_BODY|REQUEST_COOKIES \
"(?i)(;\s*(ls|cat|id|whoami|uname|pwd|wget|curl|nc|bash|sh|python|perl|ruby)\s|\$\(|`|&&\s*\w+\s|\|\|\s*\w+\s|\|\s*(ls|cat|id|whoami|uname|pwd|wget|curl|nc|bash|sh|python|perl|ruby)\s|/bin/(ba)?sh|/bin/cat|/usr/bin/(wget|curl|python|perl))" \
"id:1006,phase:2,log,deny,status:403,msg:'Command Injection attempt',tag:'attack-rce',severity:CRITICAL"

View File

@@ -0,0 +1,60 @@
syntax = "proto3";
package next_edu_cloud.ai.v1;
service AiService {
rpc Chat(ChatRequest) returns (ChatResponse);
rpc StreamChat(ChatRequest) returns (stream ChatChunk);
rpc GenerateQuestion(GenerateQuestionRequest) returns (GeneratedQuestion);
rpc OptimizeExpression(OptimizeExpressionRequest) returns (OptimizedExpression);
}
message ChatRequest {
repeated ChatMessage messages = 1;
string model = 2;
double temperature = 3;
}
message ChatMessage {
string role = 1;
string content = 2;
}
message ChatResponse {
string content = 1;
string model = 2;
Usage usage = 3;
}
message Usage {
int32 prompt_tokens = 1;
int32 completion_tokens = 2;
int32 total_tokens = 3;
}
message ChatChunk {
string content = 1;
bool done = 2;
}
message GenerateQuestionRequest {
string prompt = 1;
string subject = 2;
string difficulty = 3;
}
message GeneratedQuestion {
string question = 1;
string answer = 2;
string explanation = 3;
}
message OptimizeExpressionRequest {
string text = 1;
string context = 2;
}
message OptimizedExpression {
string optimized = 1;
repeated string suggestions = 2;
}

View File

@@ -0,0 +1,48 @@
syntax = "proto3";
package next_edu_cloud.msg.v1;
service NotificationService {
rpc SendNotification(SendNotificationRequest) returns (Notification);
rpc ListNotifications(ListNotificationsRequest) returns (ListNotificationsResponse);
rpc MarkAsRead(MarkAsReadRequest) returns (Empty);
rpc SearchNotifications(SearchNotificationsRequest) returns (SearchNotificationsResponse);
}
message Notification {
string id = 1;
string user_id = 2;
string type = 3;
string title = 4;
string content = 5;
string channel = 6;
bool is_read = 7;
int64 created_at = 8;
}
message SendNotificationRequest {
string user_id = 1;
string type = 2;
string title = 3;
string content = 4;
string channel = 5;
}
message ListNotificationsRequest {
string user_id = 1;
bool only_unread = 2;
}
message ListNotificationsResponse {
repeated Notification notifications = 1;
}
message MarkAsReadRequest { string id = 1; }
message SearchNotificationsRequest {
string user_id = 1;
string query = 2;
}
message SearchNotificationsResponse {
repeated Notification notifications = 1;
}
message Empty {}

8
services/ai/Dockerfile Normal file
View File

@@ -0,0 +1,8 @@
FROM python:3.12-slim
WORKDIR /app
RUN pip install uv
COPY pyproject.toml .
RUN uv sync --no-dev
COPY src ./src
EXPOSE 3008
CMD ["uv", "run", "uvicorn", "src.ai.main:app", "--host", "0.0.0.0", "--port", "3008"]

45
services/ai/README.md Normal file
View File

@@ -0,0 +1,45 @@
# AI 网关服务
> 版本0.1P5 骨架)
> 端口3008
## 职责
AI 网关限界上下文Python 实现),统一封装 LLM 调用(多模型路由、重试、限流、成本控制)。
提供辅助出题、表达优化、分层提问等能力。通过 gRPC 查询 content 题库与 data-ana 学情数据。
## 技术栈
- Python 3.12 + FastAPI 0.115
- Pydantic 2 + pydantic-settings
- OpenTelemetryLLM 调用链追踪)
- prometheus-client + structlog
- SSE 流式响应
## 开发
```bash
uv sync
uv run uvicorn src.ai.main:app --reload --port 3008
```
## API
| 方法 | 路径 | 说明 |
|------|------|------|
| GET | /healthz | 健康检查 |
| POST | /chat | LLM 聊天接口 |
| POST | /chat/stream | 流式聊天SSE |
| POST | /generate/question | 生成题目 |
| POST | /optimize/expression | 优化表达 |
| GET | /metrics | Prometheus 指标 |
## 环境变量
| 变量 | 默认值 | 说明 |
|------|--------|------|
| port | 3008 | 服务端口 |
| openai_api_key | - | OpenAI API 密钥 |
| anthropic_api_key | - | Anthropic API 密钥 |
| otel_endpoint | http://localhost:4318 | OpenTelemetry OTLP 端点 |
| log_level | info | 日志级别 |

View File

@@ -0,0 +1,23 @@
[project]
name = "ai-service"
version = "0.1.0"
description = "AI 网关服务 - LLM 集成 + RAG"
requires-python = ">=3.12"
dependencies = [
"fastapi>=0.115.0",
"uvicorn[standard]>=0.30.0",
"pydantic>=2.9.0",
"pydantic-settings>=2.5.0",
"httpx>=0.27.0",
"opentelemetry-api>=1.27.0",
"opentelemetry-sdk>=1.27.0",
"prometheus-client>=0.20.0",
"structlog>=24.4.0",
]
[tool.ruff]
line-length = 100
target-version = "py312"
[tool.ruff.lint]
select = ["E", "F", "I", "N", "W", "UP", "B", "SIM"]

View File

View File

@@ -0,0 +1,18 @@
"""配置管理."""
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
"""应用配置."""
port: int = 3008
openai_api_key: str = ""
anthropic_api_key: str = ""
otel_endpoint: str = "http://localhost:4318"
log_level: str = "info"
model_config = {"env_file": ".env", "env_prefix": ""}
settings = Settings()

111
services/ai/src/ai/main.py Normal file
View File

@@ -0,0 +1,111 @@
"""AI 网关服务入口."""
from contextlib import asynccontextmanager
import structlog
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from prometheus_client import make_asgi_app
from pydantic import BaseModel
logger = structlog.get_logger()
tracer = trace.get_tracer(__name__)
def init_tracer() -> None:
"""初始化 OpenTelemetry."""
provider = TracerProvider()
exporter = OTLPSpanExporter(endpoint="http://localhost:4318/v1/traces")
provider.add_span_processor(BatchSpanProcessor(exporter))
trace.set_tracer_provider(provider)
@asynccontextmanager
async def lifespan(app: FastAPI):
"""应用生命周期."""
init_tracer()
logger.info("ai service starting")
yield
logger.info("ai service stopping")
app = FastAPI(
title="AI Gateway Service",
version="0.1.0",
lifespan=lifespan,
)
app.mount("/metrics", make_asgi_app())
class ChatRequest(BaseModel):
"""聊天请求."""
messages: list[dict]
model: str = "gpt-4o-mini"
temperature: float = 0.7
stream: bool = False
class ChatResponse(BaseModel):
"""聊天响应."""
content: str
model: str
usage: dict
@app.get("/healthz")
async def healthz():
"""健康检查."""
return {"status": "ok", "service": "ai"}
@app.post("/chat", response_model=ChatResponse)
async def chat(req: ChatRequest):
"""LLM 聊天接口."""
with tracer.start_as_current_span("ai_chat"):
# P5 骨架:实际调用 OpenAI/Anthropic API
# 需要从环境变量获取 API key
return {
"content": "P5 skeleton - LLM integration pending",
"model": req.model,
"usage": {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0},
}
@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
"""流式聊天SSE."""
async def generate():
with tracer.start_as_current_span("ai_chat_stream"):
# P5 骨架:流式调用 LLM
yield "data: P5 skeleton\n\n"
yield "data: [DONE]\n\n"
return StreamingResponse(generate(), media_type="text/event-stream")
@app.post("/generate/question")
async def generate_question(prompt: str):
"""生成题目."""
with tracer.start_as_current_span("generate_question"):
return {
"success": True,
"data": {"question": "P5 skeleton - question generation pending"},
}
@app.post("/optimize/expression")
async def optimize_expression(text: str):
"""优化表达."""
with tracer.start_as_current_span("optimize_expression"):
return {
"success": True,
"data": {"optimized": "P5 skeleton - expression optimization pending"},
}

View File

@@ -0,0 +1,36 @@
"""健康检查端点ai 服务)。
- GET /healthzliveness仅返回进程存活。
- GET /readyzreadiness简化版返回 ok + TODO待补全 Elasticsearch
连通性与 OpenAI 配置校验。
集成说明:在 FastAPI app 中挂载 router
from health import router as health_router
app.include_router(health_router)
"""
from datetime import datetime, timezone
from fastapi import APIRouter
router = APIRouter()
SERVICE_NAME = "ai"
@router.get("/healthz")
async def healthz() -> dict:
return {"status": "ok", "service": SERVICE_NAME}
@router.get("/readyz")
async def readyz() -> dict:
# TODO: 校验关键依赖
# 1. Elasticsearch 连通性es_client.ping()
# 2. OpenAI 配置API key 是否注入、超时是否合理
# 依赖客户端就绪后再补全检查逻辑,失败时返回 503。
return {
"status": "ok",
"service": SERVICE_NAME,
"timestamp": datetime.now(timezone.utc).isoformat(),
}

View File

@@ -6,6 +6,7 @@ require (
github.com/gin-gonic/gin v1.10.0
github.com/golang-jwt/jwt/v5 v5.2.1
github.com/google/uuid v1.6.0
github.com/sony/gobreaker/v2 v2.1.0
go.opentelemetry.io/contrib/instrumentation/github.com/gin-gonic/gin/otelgin v0.52.0
go.opentelemetry.io/otel v1.27.0
go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.27.0

View File

@@ -0,0 +1,23 @@
package health
import (
"github.com/gin-gonic/gin"
)
// Healthz 存活探针liveness
// GET /healthz返回 200 {"status":"ok"} 表示进程存活。
func Healthz(c *gin.Context) {
c.JSON(200, gin.H{
"status": "ok",
})
}
// Readyz 就绪探针readiness
// GET /readyz检查下游服务可达性。
// TODO: P7 接入服务发现后,检查 classes / iam / teacher-bff 等下游服务健康状态,
// 任一不可达则返回 503当前简化为直接返回 200。
func Readyz(c *gin.Context) {
c.JSON(200, gin.H{
"status": "ok",
})
}

View File

@@ -0,0 +1,65 @@
package middleware
import (
"errors"
"log"
"net/http"
"time"
"github.com/gin-gonic/gin"
"github.com/sony/gobreaker/v2"
)
// CircuitBreaker 返回针对指定服务的熔断中间件。
//
// 基于 gobreaker v2 实现:
// - Interval=5sCLOSED 状态下的统计窗口
// - ReadyToTrip错误率 > 50% 触发 OPEN
// - Timeout=30sOPEN 持续 30 秒后转 HALF_OPEN
// - MaxRequests=1HALF_OPEN 仅允许 1 个探测请求
//
// 仅当下游返回 5xx 视为失败4xx 与 2xx 不计入熔断。
// 熔断打开或半开探测名额已满时返回 503 + JSON {"error":"circuit_open","retry_after":30}。
func CircuitBreaker(serviceName string) gin.HandlerFunc {
cb := gobreaker.NewCircuitBreaker[struct{}](gobreaker.Settings{
Name: serviceName,
MaxRequests: 1,
Interval: 5 * time.Second,
Timeout: 30 * time.Second,
ReadyToTrip: func(counts gobreaker.Counts) bool {
// 请求数为 0 时不触发,避免除零
if counts.Requests == 0 {
return false
}
// 错误率 > 50%
return counts.TotalFailures*2 > counts.Requests
},
OnStateChange: func(name string, from, to gobreaker.State) {
log.Printf("[circuit-breaker] service=%s state: %s -> %s", name, from, to)
},
})
return func(c *gin.Context) {
_, err := cb.Execute(func() (struct{}, error) {
c.Next()
// 下游 5xx 视为熔断失败
if c.Writer.Status() >= 500 {
return struct{}{}, errors.New("downstream_error")
}
return struct{}{}, nil
})
if err != nil {
// 熔断打开或半开探测名额已满:返回 503
if errors.Is(err, gobreaker.ErrOpenState) || errors.Is(err, gobreaker.ErrTooManyRequests) {
c.AbortWithStatusJSON(http.StatusServiceUnavailable, gin.H{
"error": "circuit_open",
"retry_after": 30,
})
return
}
// 其他情况(下游已写 5xx 响应):响应已写入,不覆盖
return
}
}
}

View File

@@ -0,0 +1,69 @@
package middleware
import (
"net/http"
"os"
"strconv"
"strings"
"github.com/gin-gonic/gin"
)
// corsMaxAge 预检缓存时长12 小时
const corsMaxAge = 12 * 60 * 60
// CORS 返回跨域资源共享中间件。
// 允许来源从环境变量 CORS_ORIGINS 读取(逗号分隔,默认 *)。
// 允许方法GET POST PUT DELETE OPTIONS PATCH
// 允许头Authorization Content-Type X-Request-Id X-Trace-Id
// 暴露头X-Request-Id X-Trace-Id
func CORS() gin.HandlerFunc {
allowed := parseCORSOrigins(os.Getenv("CORS_ORIGINS"))
return func(c *gin.Context) {
origin := c.GetHeader("Origin")
allowOrigin := ""
if len(allowed) == 0 {
// 未配置则默认允许所有来源
allowOrigin = "*"
} else if allowed[origin] {
allowOrigin = origin
}
if allowOrigin != "" {
h := c.Writer.Header()
h.Set("Access-Control-Allow-Origin", allowOrigin)
h.Set("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS, PATCH")
h.Set("Access-Control-Allow-Headers", "Authorization, Content-Type, X-Request-Id, X-Trace-Id")
h.Set("Access-Control-Expose-Headers", "X-Request-Id, X-Trace-Id")
h.Set("Access-Control-Max-Age", strconv.Itoa(corsMaxAge))
// 非通配来源需标注 Vary便于缓存正确区分
if allowOrigin != "*" {
h.Add("Vary", "Origin")
}
}
// 预检请求直接返回 204
if c.Request.Method == http.MethodOptions {
c.AbortWithStatus(http.StatusNoContent)
return
}
c.Next()
}
}
// parseCORSOrigins 解析逗号分隔的来源列表为集合,空字符串返回空 map表示通配 *
func parseCORSOrigins(raw string) map[string]bool {
allowed := map[string]bool{}
if raw == "" {
return allowed
}
for _, o := range strings.Split(raw, ",") {
o = strings.TrimSpace(o)
if o != "" {
allowed[o] = true
}
}
return allowed
}

View File

@@ -0,0 +1,94 @@
package middleware
import (
"net/http"
"sync"
"time"
"github.com/gin-gonic/gin"
)
// bucket 是单个客户端的令牌桶
type bucket struct {
mu sync.Mutex
tokens float64
lastTime time.Time
}
// rateLimiter 持有所有客户端 IP 的令牌桶
type rateLimiter struct {
rps float64
burst int
buckets sync.Map // map[string]*bucket
}
// RateLimit 返回基于令牌桶的限流中间件(内存版,不依赖 Redis
// 每个客户端 IP 独立桶,按 rps每秒令牌数补充最大容量为 burst。
// 超限返回 429 + JSON {"error":"rate_limited","retry_after":60}。
// 内部启动 cleanup goroutine每 5 分钟清理 10 分钟未访问的桶。
func RateLimit(rps float64, burst int) gin.HandlerFunc {
rl := &rateLimiter{rps: rps, burst: burst}
// 启动清理 goroutine
go rl.cleanup(5*time.Minute, 10*time.Minute)
return func(c *gin.Context) {
ip := c.ClientIP()
b := rl.getBucket(ip)
b.mu.Lock()
now := time.Now()
elapsed := now.Sub(b.lastTime).Seconds()
b.lastTime = now
// 按经过时间补充令牌
b.tokens += elapsed * rl.rps
if b.tokens > float64(rl.burst) {
b.tokens = float64(rl.burst)
}
// 令牌不足:拒绝
if b.tokens < 1 {
b.mu.Unlock()
c.Header("Retry-After", "60")
c.AbortWithStatusJSON(http.StatusTooManyRequests, gin.H{
"error": "rate_limited",
"retry_after": 60,
})
return
}
b.tokens--
b.mu.Unlock()
c.Next()
}
}
// getBucket 获取指定 IP 的令牌桶,不存在则新建(初始满桶)
func (rl *rateLimiter) getBucket(ip string) *bucket {
if v, ok := rl.buckets.Load(ip); ok {
return v.(*bucket)
}
b := &bucket{
tokens: float64(rl.burst),
lastTime: time.Now(),
}
actual, _ := rl.buckets.LoadOrStore(ip, b)
return actual.(*bucket)
}
// cleanup 周期性清理长时间未访问的桶,避免内存无限增长
func (rl *rateLimiter) cleanup(interval, maxIdle time.Duration) {
ticker := time.NewTicker(interval)
defer ticker.Stop()
for range ticker.C {
now := time.Now()
rl.buckets.Range(func(key, value any) bool {
b := value.(*bucket)
b.mu.Lock()
idle := now.Sub(b.lastTime)
b.mu.Unlock()
if idle > maxIdle {
rl.buckets.Delete(key)
}
return true
})
}
}

View File

@@ -0,0 +1,32 @@
package middleware
import (
"log"
"net/http"
"runtime/debug"
"github.com/gin-gonic/gin"
"github.com/google/uuid"
)
// Recovery 捕获 panic 并返回 500记录堆栈日志。
// 返回 JSON {"error":"internal_error","request_id":"<uuid>"}。
// request_id 使用 uuid.New() 生成Recovery 在 RequestID 之前注册,
// panic 发生时上下文中可能尚无 request_id故独立生成
func Recovery() gin.HandlerFunc {
return func(c *gin.Context) {
defer func() {
if r := recover(); r != nil {
stack := debug.Stack()
requestID := uuid.New().String()
log.Printf("[recovery] panic recovered, request_id=%s: %v\n%s", requestID, r, stack)
// 若已写入部分响应AbortWithStatusJSON 仍会设置状态并尝试写 JSON
c.AbortWithStatusJSON(http.StatusInternalServerError, gin.H{
"error": "internal_error",
"request_id": requestID,
})
}
}()
c.Next()
}
}

View File

@@ -0,0 +1,26 @@
package middleware
import (
"github.com/gin-gonic/gin"
"github.com/google/uuid"
)
// requestIDHeader 请求 ID 响应头名称
const requestIDHeader = "X-Request-Id"
// requestIDContextKey context 中存储请求 ID 的键
const requestIDContextKey = "request_id"
// RequestID 注入请求 ID 中间件。
// 从 X-Request-Id 头读取,无则生成 uuid写入 context 与响应头。
func RequestID() gin.HandlerFunc {
return func(c *gin.Context) {
rid := c.GetHeader(requestIDHeader)
if rid == "" {
rid = uuid.New().String()
}
c.Set(requestIDContextKey, rid)
c.Writer.Header().Set(requestIDHeader, rid)
c.Next()
}
}

View File

@@ -0,0 +1,39 @@
package middleware
import (
"net/http"
"github.com/gin-gonic/gin"
)
// SecurityHeaders 设置安全响应头中间件。
// 设置X-Content-Type-Options: nosniff
//
// X-Frame-Options: DENY
// Referrer-Policy: strict-origin-when-cross-origin
// Strict-Transport-Security: max-age=31536000; includeSubDomains
// Content-Security-Policy: default-src 'self'
// 移除 Server 头
func SecurityHeaders() gin.HandlerFunc {
return func(c *gin.Context) {
h := c.Writer.Header()
h.Set("X-Content-Type-Options", "nosniff")
h.Set("X-Frame-Options", "DENY")
h.Set("Referrer-Policy", "strict-origin-when-cross-origin")
h.Set("Strict-Transport-Security", "max-age=31536000; includeSubDomains")
h.Set("Content-Security-Policy", "default-src 'self'")
h.Del("Server")
c.Next()
// 响应处理结束后再次移除 Server 头,防止处理过程中被写入
c.Writer.Header().Del("Server")
}
}
// RequestBodyLimit 限制请求体大小中间件。
// 通过 http.MaxBytesReader 包装 Body超限读取时返回 413。
func RequestBodyLimit(maxBytes int64) gin.HandlerFunc {
return func(c *gin.Context) {
c.Request.Body = http.MaxBytesReader(c.Writer, c.Request.Body, maxBytes)
c.Next()
}
}

View File

@@ -10,13 +10,66 @@ import (
"time"
"github.com/edu-cloud/api-gateway/internal/config"
"github.com/edu-cloud/api-gateway/internal/routing"
"github.com/edu-cloud/api-gateway/internal/health"
"github.com/edu-cloud/api-gateway/internal/middleware"
"github.com/edu-cloud/api-gateway/internal/proxy"
"github.com/gin-gonic/gin"
)
// maxBodyBytes 请求体大小上限10MB
const maxBodyBytes int64 = 10 * 1024 * 1024
func main() {
cfg := config.Load()
gin.SetMode(gin.ReleaseMode)
r := gin.New()
r := routing.Setup(cfg)
// 全局中间件(按顺序注册)
// 1. panic 恢复(最外层,捕获后续所有中间件与 handler 的 panic
r.Use(middleware.Recovery())
// 2. 请求 ID 注入
r.Use(middleware.RequestID())
// 3. 跨域
r.Use(middleware.CORS())
// 4. 安全响应头
r.Use(middleware.SecurityHeaders())
// 5. 请求体大小限制
r.Use(middleware.RequestBodyLimit(maxBodyBytes))
// 6. 限流(每 IP 100 rps突发 20
r.Use(middleware.RateLimit(100, 20))
// 健康检查路由(无需鉴权,在 Auth 之前)
r.GET("/healthz", health.Healthz)
r.GET("/readyz", health.Readyz)
// API v1 组:熔断 + 鉴权 + 反向代理
api := r.Group("/api/v1")
// 7. 熔断(仅作用于代理路由,下游 5xx 触发)
api.Use(middleware.CircuitBreaker("downstream"))
// 8. JWT 鉴权
api.Use(middleware.AuthMiddleware(cfg))
{
// classes 服务路由
classesProxy, err := proxy.NewProxy(cfg.ClassesServiceURL)
if err != nil {
log.Fatalf("failed to create classes proxy: %v", err)
}
api.Any("/classes/*path", proxy.ProxyHandler(classesProxy))
// IAM 服务路由(身份与访问管理)
iamProxy, err := proxy.NewProxy(cfg.IamServiceURL)
if err != nil {
log.Fatalf("failed to create iam proxy: %v", err)
}
api.Any("/iam/*path", proxy.ProxyHandler(iamProxy))
// Teacher BFF 路由(教师聚合层)
bffProxy, err := proxy.NewProxy(cfg.TeacherBffURL)
if err != nil {
log.Fatalf("failed to create teacher-bff proxy: %v", err)
}
api.Any("/teacher/*path", proxy.ProxyHandler(bffProxy))
}
srv := &http.Server{
Addr: ":" + cfg.Port,
@@ -29,7 +82,7 @@ func main() {
go func() {
log.Printf("API Gateway listening on :%s", cfg.Port)
if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
log.Fatalf("listen: %s\n", err)
log.Fatalf("listen: %s", err)
}
}()

View File

@@ -0,0 +1,49 @@
import { Controller, Get, HttpException, HttpStatus } from '@nestjs/common';
import { DataSource } from 'typeorm';
const SERVICE_NAME = 'classes';
/**
* 健康检查端点。
*
* - GET /healthzliveness仅返回进程存活不检查依赖。
* - GET /readyzreadiness检查 DB 连接,失败返回 503。
*
* 不需要鉴权,必须在路由白名单中放行。本控制器内容在 iam / core-edu /
* content / msg / classes 五个 NestJS 服务中一致,仅 SERVICE_NAME 不同。
*/
@Controller()
export class HealthController {
constructor(private readonly dataSource: DataSource) {}
@Get('healthz')
liveness(): { status: string; service: string; timestamp: string } {
return {
status: 'ok',
service: SERVICE_NAME,
timestamp: new Date().toISOString(),
};
}
@Get('readyz')
async readiness(): Promise<{ status: string; service: string; timestamp: string }> {
try {
await this.dataSource.query('SELECT 1');
return {
status: 'ok',
service: SERVICE_NAME,
timestamp: new Date().toISOString(),
};
} catch (error) {
throw new HttpException(
{
status: 'error',
service: SERVICE_NAME,
timestamp: new Date().toISOString(),
error: error instanceof Error ? error.message : 'database unreachable',
},
HttpStatus.SERVICE_UNAVAILABLE,
);
}
}
}

View File

@@ -0,0 +1,23 @@
import { Module } from '@nestjs/common';
import { HealthController } from './health.controller';
/**
* 健康检查模块。
*
* 集成说明(不修改 app.module.ts仅在 README 注释说明):
*
* 在 `app.module.ts` 的 imports 数组中加入 `HealthModule`
*
* ```ts
* import { HealthModule } from './shared/health/health.module';
*
* @Module({ imports: [ ..., HealthModule ], ... })
* export class AppModule {}
* ```
*
* DataSource 由 `TypeOrmModule.forRoot(...)` 提供,本模块无需额外 provider。
*/
@Module({
controllers: [HealthController],
})
export class HealthModule {}

View File

@@ -0,0 +1,63 @@
import { Inject, Injectable, Logger, OnApplicationShutdown, OnModuleInit } from '@nestjs/common';
import { DataSource } from 'typeorm';
import type { Redis } from 'ioredis';
import type { Producer } from 'kafkajs';
const SERVICE_NAME = 'classes';
/**
* 优雅停机服务。
*
* 信号处理由 NestJS 在 `app.listen` 之前调用 `app.enableShutdownHooks()`
* 触发SIGTERM / SIGINTNestJS 会依次调用 OnApplicationShutdown 钩子。
* K8s 配置 `terminationGracePeriodSeconds=60` 给予足够时间清理。
*
* 关闭顺序Kafka producer → Redis → DataSource。
* 先停外部消息生产避免新事件,再关缓存,最后关 DB。
*
* 集成说明(不修改 app.module.ts仅在 README 注释说明):
* - 在 `app.module.ts` 的 providers 中加入 `LifecycleService`。
* - 在 `main.ts` 中 `app.listen` 之前调用 `app.enableShutdownHooks()`。
*
* 依赖注入 token 约定(需与各服务 provider 注册一致):
* - DataSource由 `TypeOrmModule.forRoot()` 提供。
* - 'REDIS_CLIENT':需在对应模块注册 `{ provide: 'REDIS_CLIENT', useFactory: ... }`。
* - 'KAFKA_PRODUCER':需在对应模块注册 `{ provide: 'KAFKA_PRODUCER', useFactory: ... }`。
*/
@Injectable()
export class LifecycleService implements OnModuleInit, OnApplicationShutdown {
private readonly logger = new Logger(LifecycleService.name);
constructor(
private readonly dataSource: DataSource,
@Inject('REDIS_CLIENT') private readonly redis: Redis,
@Inject('KAFKA_PRODUCER') private readonly kafkaProducer: Producer,
) {}
onModuleInit(): void {
this.logger.log(`service ${SERVICE_NAME} module initialized`);
}
async onApplicationShutdown(signal?: string): Promise<void> {
this.logger.log(
`service ${SERVICE_NAME} shutting down (signal=${signal ?? 'unknown'})`,
);
await this.safeDisconnect('kafka producer', () => this.kafkaProducer.disconnect());
await this.safeDisconnect('redis', () => this.redis.quit());
await this.safeDisconnect('datasource', () => this.dataSource.destroy());
this.logger.log(`service ${SERVICE_NAME} shutdown complete`);
}
private async safeDisconnect(name: string, fn: () => Promise<unknown>): Promise<void> {
try {
await fn();
this.logger.log(`${name} closed`);
} catch (error) {
this.logger.error(
`${name} close failed: ${error instanceof Error ? error.message : String(error)}`,
);
}
}
}

View File

@@ -0,0 +1,49 @@
import { Controller, Get, HttpException, HttpStatus } from '@nestjs/common';
import { DataSource } from 'typeorm';
const SERVICE_NAME = 'content';
/**
* 健康检查端点。
*
* - GET /healthzliveness仅返回进程存活不检查依赖。
* - GET /readyzreadiness检查 DB 连接,失败返回 503。
*
* 不需要鉴权,必须在路由白名单中放行。本控制器内容在 iam / core-edu /
* content / msg / classes 五个 NestJS 服务中一致,仅 SERVICE_NAME 不同。
*/
@Controller()
export class HealthController {
constructor(private readonly dataSource: DataSource) {}
@Get('healthz')
liveness(): { status: string; service: string; timestamp: string } {
return {
status: 'ok',
service: SERVICE_NAME,
timestamp: new Date().toISOString(),
};
}
@Get('readyz')
async readiness(): Promise<{ status: string; service: string; timestamp: string }> {
try {
await this.dataSource.query('SELECT 1');
return {
status: 'ok',
service: SERVICE_NAME,
timestamp: new Date().toISOString(),
};
} catch (error) {
throw new HttpException(
{
status: 'error',
service: SERVICE_NAME,
timestamp: new Date().toISOString(),
error: error instanceof Error ? error.message : 'database unreachable',
},
HttpStatus.SERVICE_UNAVAILABLE,
);
}
}
}

View File

@@ -0,0 +1,23 @@
import { Module } from '@nestjs/common';
import { HealthController } from './health.controller';
/**
* 健康检查模块。
*
* 集成说明(不修改 app.module.ts仅在 README 注释说明):
*
* 在 `app.module.ts` 的 imports 数组中加入 `HealthModule`
*
* ```ts
* import { HealthModule } from './shared/health/health.module';
*
* @Module({ imports: [ ..., HealthModule ], ... })
* export class AppModule {}
* ```
*
* DataSource 由 `TypeOrmModule.forRoot(...)` 提供,本模块无需额外 provider。
*/
@Module({
controllers: [HealthController],
})
export class HealthModule {}

View File

@@ -0,0 +1,64 @@
import { Inject, Injectable, Logger, OnApplicationShutdown, OnModuleInit } from '@nestjs/common';
import { DataSource } from 'typeorm';
import type { Redis } from 'ioredis';
import type { Producer } from 'kafkajs';
const SERVICE_NAME = 'content';
/**
* 优雅停机服务。
*
* 信号处理由 NestJS 在 `app.listen` 之前调用 `app.enableShutdownHooks()`
* 触发SIGTERM / SIGINTNestJS 会依次调用 OnApplicationShutdown 钩子。
* K8s 配置 `terminationGracePeriodSeconds=60` 给予足够时间清理。
*
* 关闭顺序Kafka producer → Redis → DataSource。
* 先停外部消息生产避免新事件,再关缓存,最后关 DB。
* 注content 服务额外使用 S3/MinIO 客户端,其连接由 SDK 内部管理,无需显式关闭。
*
* 集成说明(不修改 app.module.ts仅在 README 注释说明):
* - 在 `app.module.ts` 的 providers 中加入 `LifecycleService`。
* - 在 `main.ts` 中 `app.listen` 之前调用 `app.enableShutdownHooks()`。
*
* 依赖注入 token 约定(需与各服务 provider 注册一致):
* - DataSource由 `TypeOrmModule.forRoot()` 提供。
* - 'REDIS_CLIENT':需在对应模块注册 `{ provide: 'REDIS_CLIENT', useFactory: ... }`。
* - 'KAFKA_PRODUCER':需在对应模块注册 `{ provide: 'KAFKA_PRODUCER', useFactory: ... }`。
*/
@Injectable()
export class LifecycleService implements OnModuleInit, OnApplicationShutdown {
private readonly logger = new Logger(LifecycleService.name);
constructor(
private readonly dataSource: DataSource,
@Inject('REDIS_CLIENT') private readonly redis: Redis,
@Inject('KAFKA_PRODUCER') private readonly kafkaProducer: Producer,
) {}
onModuleInit(): void {
this.logger.log(`service ${SERVICE_NAME} module initialized`);
}
async onApplicationShutdown(signal?: string): Promise<void> {
this.logger.log(
`service ${SERVICE_NAME} shutting down (signal=${signal ?? 'unknown'})`,
);
await this.safeDisconnect('kafka producer', () => this.kafkaProducer.disconnect());
await this.safeDisconnect('redis', () => this.redis.quit());
await this.safeDisconnect('datasource', () => this.dataSource.destroy());
this.logger.log(`service ${SERVICE_NAME} shutdown complete`);
}
private async safeDisconnect(name: string, fn: () => Promise<unknown>): Promise<void> {
try {
await fn();
this.logger.log(`${name} closed`);
} catch (error) {
this.logger.error(
`${name} close failed: ${error instanceof Error ? error.message : String(error)}`,
);
}
}
}

View File

@@ -0,0 +1,49 @@
import { Controller, Get, HttpException, HttpStatus } from '@nestjs/common';
import { DataSource } from 'typeorm';
const SERVICE_NAME = 'core-edu';
/**
* 健康检查端点。
*
* - GET /healthzliveness仅返回进程存活不检查依赖。
* - GET /readyzreadiness检查 DB 连接,失败返回 503。
*
* 不需要鉴权,必须在路由白名单中放行。本控制器内容在 iam / core-edu /
* content / msg / classes 五个 NestJS 服务中一致,仅 SERVICE_NAME 不同。
*/
@Controller()
export class HealthController {
constructor(private readonly dataSource: DataSource) {}
@Get('healthz')
liveness(): { status: string; service: string; timestamp: string } {
return {
status: 'ok',
service: SERVICE_NAME,
timestamp: new Date().toISOString(),
};
}
@Get('readyz')
async readiness(): Promise<{ status: string; service: string; timestamp: string }> {
try {
await this.dataSource.query('SELECT 1');
return {
status: 'ok',
service: SERVICE_NAME,
timestamp: new Date().toISOString(),
};
} catch (error) {
throw new HttpException(
{
status: 'error',
service: SERVICE_NAME,
timestamp: new Date().toISOString(),
error: error instanceof Error ? error.message : 'database unreachable',
},
HttpStatus.SERVICE_UNAVAILABLE,
);
}
}
}

View File

@@ -0,0 +1,23 @@
import { Module } from '@nestjs/common';
import { HealthController } from './health.controller';
/**
* 健康检查模块。
*
* 集成说明(不修改 app.module.ts仅在 README 注释说明):
*
* 在 `app.module.ts` 的 imports 数组中加入 `HealthModule`
*
* ```ts
* import { HealthModule } from './shared/health/health.module';
*
* @Module({ imports: [ ..., HealthModule ], ... })
* export class AppModule {}
* ```
*
* DataSource 由 `TypeOrmModule.forRoot(...)` 提供,本模块无需额外 provider。
*/
@Module({
controllers: [HealthController],
})
export class HealthModule {}

View File

@@ -0,0 +1,63 @@
import { Inject, Injectable, Logger, OnApplicationShutdown, OnModuleInit } from '@nestjs/common';
import { DataSource } from 'typeorm';
import type { Redis } from 'ioredis';
import type { Producer } from 'kafkajs';
const SERVICE_NAME = 'core-edu';
/**
* 优雅停机服务。
*
* 信号处理由 NestJS 在 `app.listen` 之前调用 `app.enableShutdownHooks()`
* 触发SIGTERM / SIGINTNestJS 会依次调用 OnApplicationShutdown 钩子。
* K8s 配置 `terminationGracePeriodSeconds=60` 给予足够时间清理。
*
* 关闭顺序Kafka producer → Redis → DataSource。
* 先停外部消息生产避免新事件,再关缓存,最后关 DB。
*
* 集成说明(不修改 app.module.ts仅在 README 注释说明):
* - 在 `app.module.ts` 的 providers 中加入 `LifecycleService`。
* - 在 `main.ts` 中 `app.listen` 之前调用 `app.enableShutdownHooks()`。
*
* 依赖注入 token 约定(需与各服务 provider 注册一致):
* - DataSource由 `TypeOrmModule.forRoot()` 提供。
* - 'REDIS_CLIENT':需在对应模块注册 `{ provide: 'REDIS_CLIENT', useFactory: ... }`。
* - 'KAFKA_PRODUCER':需在对应模块注册 `{ provide: 'KAFKA_PRODUCER', useFactory: ... }`。
*/
@Injectable()
export class LifecycleService implements OnModuleInit, OnApplicationShutdown {
private readonly logger = new Logger(LifecycleService.name);
constructor(
private readonly dataSource: DataSource,
@Inject('REDIS_CLIENT') private readonly redis: Redis,
@Inject('KAFKA_PRODUCER') private readonly kafkaProducer: Producer,
) {}
onModuleInit(): void {
this.logger.log(`service ${SERVICE_NAME} module initialized`);
}
async onApplicationShutdown(signal?: string): Promise<void> {
this.logger.log(
`service ${SERVICE_NAME} shutting down (signal=${signal ?? 'unknown'})`,
);
await this.safeDisconnect('kafka producer', () => this.kafkaProducer.disconnect());
await this.safeDisconnect('redis', () => this.redis.quit());
await this.safeDisconnect('datasource', () => this.dataSource.destroy());
this.logger.log(`service ${SERVICE_NAME} shutdown complete`);
}
private async safeDisconnect(name: string, fn: () => Promise<unknown>): Promise<void> {
try {
await fn();
this.logger.log(`${name} closed`);
} catch (error) {
this.logger.error(
`${name} close failed: ${error instanceof Error ? error.message : String(error)}`,
);
}
}
}

View File

@@ -0,0 +1,34 @@
"""健康检查端点data-ana 服务)。
- GET /healthzliveness仅返回进程存活。
- GET /readyzreadiness简化版返回 ok + TODO待补全 ClickHouse 连通性校验。
集成说明:在 FastAPI app 中挂载 router
from health import router as health_router
app.include_router(health_router)
"""
from datetime import datetime, timezone
from fastapi import APIRouter
router = APIRouter()
SERVICE_NAME = "data-ana"
@router.get("/healthz")
async def healthz() -> dict:
return {"status": "ok", "service": SERVICE_NAME}
@router.get("/readyz")
async def readyz() -> dict:
# TODO: 校验关键依赖
# 1. ClickHouse 连通性clickhouse_client.execute("SELECT 1")
# 依赖客户端就绪后再补全检查逻辑,失败时返回 503。
return {
"status": "ok",
"service": SERVICE_NAME,
"timestamp": datetime.now(timezone.utc).isoformat(),
}

View File

@@ -0,0 +1,49 @@
import { Controller, Get, HttpException, HttpStatus } from '@nestjs/common';
import { DataSource } from 'typeorm';
const SERVICE_NAME = 'iam';
/**
* 健康检查端点。
*
* - GET /healthzliveness仅返回进程存活不检查依赖。
* - GET /readyzreadiness检查 DB 连接,失败返回 503。
*
* 不需要鉴权,必须在路由白名单中放行。本控制器内容在 iam / core-edu /
* content / msg / classes 五个 NestJS 服务中一致,仅 SERVICE_NAME 不同。
*/
@Controller()
export class HealthController {
constructor(private readonly dataSource: DataSource) {}
@Get('healthz')
liveness(): { status: string; service: string; timestamp: string } {
return {
status: 'ok',
service: SERVICE_NAME,
timestamp: new Date().toISOString(),
};
}
@Get('readyz')
async readiness(): Promise<{ status: string; service: string; timestamp: string }> {
try {
await this.dataSource.query('SELECT 1');
return {
status: 'ok',
service: SERVICE_NAME,
timestamp: new Date().toISOString(),
};
} catch (error) {
throw new HttpException(
{
status: 'error',
service: SERVICE_NAME,
timestamp: new Date().toISOString(),
error: error instanceof Error ? error.message : 'database unreachable',
},
HttpStatus.SERVICE_UNAVAILABLE,
);
}
}
}

View File

@@ -0,0 +1,23 @@
import { Module } from '@nestjs/common';
import { HealthController } from './health.controller';
/**
* 健康检查模块。
*
* 集成说明(不修改 app.module.ts仅在 README 注释说明):
*
* 在 `app.module.ts` 的 imports 数组中加入 `HealthModule`
*
* ```ts
* import { HealthModule } from './shared/health/health.module';
*
* @Module({ imports: [ ..., HealthModule ], ... })
* export class AppModule {}
* ```
*
* DataSource 由 `TypeOrmModule.forRoot(...)` 提供,本模块无需额外 provider。
*/
@Module({
controllers: [HealthController],
})
export class HealthModule {}

View File

@@ -0,0 +1,63 @@
import { Inject, Injectable, Logger, OnApplicationShutdown, OnModuleInit } from '@nestjs/common';
import { DataSource } from 'typeorm';
import type { Redis } from 'ioredis';
import type { Producer } from 'kafkajs';
const SERVICE_NAME = 'iam';
/**
* 优雅停机服务。
*
* 信号处理由 NestJS 在 `app.listen` 之前调用 `app.enableShutdownHooks()`
* 触发SIGTERM / SIGINTNestJS 会依次调用 OnApplicationShutdown 钩子。
* K8s 配置 `terminationGracePeriodSeconds=60` 给予足够时间清理。
*
* 关闭顺序Kafka producer → Redis → DataSource。
* 先停外部消息生产避免新事件,再关缓存,最后关 DB。
*
* 集成说明(不修改 app.module.ts仅在 README 注释说明):
* - 在 `app.module.ts` 的 providers 中加入 `LifecycleService`。
* - 在 `main.ts` 中 `app.listen` 之前调用 `app.enableShutdownHooks()`。
*
* 依赖注入 token 约定(需与各服务 provider 注册一致):
* - DataSource由 `TypeOrmModule.forRoot()` 提供。
* - 'REDIS_CLIENT':需在对应模块注册 `{ provide: 'REDIS_CLIENT', useFactory: ... }`。
* - 'KAFKA_PRODUCER':需在对应模块注册 `{ provide: 'KAFKA_PRODUCER', useFactory: ... }`。
*/
@Injectable()
export class LifecycleService implements OnModuleInit, OnApplicationShutdown {
private readonly logger = new Logger(LifecycleService.name);
constructor(
private readonly dataSource: DataSource,
@Inject('REDIS_CLIENT') private readonly redis: Redis,
@Inject('KAFKA_PRODUCER') private readonly kafkaProducer: Producer,
) {}
onModuleInit(): void {
this.logger.log(`service ${SERVICE_NAME} module initialized`);
}
async onApplicationShutdown(signal?: string): Promise<void> {
this.logger.log(
`service ${SERVICE_NAME} shutting down (signal=${signal ?? 'unknown'})`,
);
await this.safeDisconnect('kafka producer', () => this.kafkaProducer.disconnect());
await this.safeDisconnect('redis', () => this.redis.quit());
await this.safeDisconnect('datasource', () => this.dataSource.destroy());
this.logger.log(`service ${SERVICE_NAME} shutdown complete`);
}
private async safeDisconnect(name: string, fn: () => Promise<unknown>): Promise<void> {
try {
await fn();
this.logger.log(`${name} closed`);
} catch (error) {
this.logger.error(
`${name} close failed: ${error instanceof Error ? error.message : String(error)}`,
);
}
}
}

19
services/msg/Dockerfile Normal file
View File

@@ -0,0 +1,19 @@
# Build stage
FROM node:20-alpine AS builder
WORKDIR /app
RUN npm install -g pnpm
COPY package.json pnpm-lock.yaml* ./
RUN pnpm install --frozen-lockfile
COPY tsconfig.json nest-cli.json ./
COPY src ./src
RUN pnpm build
# Runtime stage
FROM node:20-alpine
WORKDIR /app
RUN npm install -g pnpm
COPY package.json pnpm-lock.yaml* ./
RUN pnpm install --prod --frozen-lockfile
COPY --from=builder /app/dist ./dist
EXPOSE 3007
CMD ["node", "dist/main.js"]

43
services/msg/README.md Normal file
View File

@@ -0,0 +1,43 @@
# Msg 消息通知服务
> 版本0.1P5 骨架)
> 端口3007
## 职责
消息通知限界上下文,管理通知的多渠道分发(站内信、邮件、短信、推送)。
消费 Kafka 事件,写入 MySQL + Elasticsearch调用 Push Gateway 实时推送。
## 技术栈
- NestJS 10 + TypeScript 5
- Drizzle ORM + MySQL 8
- Elasticsearch 8全文检索
- Kafka事件消费骨架
- pino + prom-client + OpenTelemetry
## 开发
```bash
pnpm install
pnpm dev # http://localhost:3007
```
## API
| 方法 | 路径 | 说明 |
|------|------|------|
| POST | /notifications | 发送通知 |
| GET | /notifications | 查询用户通知列表 |
| POST | /notifications/:id/read | 标记已读 |
| GET | /notifications/search?q= | 全文检索通知 |
## 环境变量
| 变量 | 默认值 | 说明 |
|------|--------|------|
| PORT | 3007 | 服务端口 |
| DATABASE_URL | - | MySQL 连接串 |
| KAFKA_BROKERS | localhost:9092 | Kafka broker 列表 |
| ES_URL | http://localhost:9200 | Elasticsearch 地址 |
| JWT_SECRET | - | JWT 密钥 |

View File

@@ -0,0 +1,8 @@
{
"$schema": "https://json.schemastore.org/nest-cli",
"collection": "@nestjs/schematics",
"sourceRoot": "src",
"compilerOptions": {
"deleteOutDir": true
}
}

38
services/msg/package.json Normal file
View File

@@ -0,0 +1,38 @@
{
"name": "@edu/msg-service",
"version": "0.1.0",
"private": true,
"type": "module",
"scripts": {
"dev": "nest start --watch",
"build": "nest build",
"start": "node dist/main.js",
"test": "vitest run",
"lint": "eslint src --ext .ts",
"typecheck": "tsc --noEmit"
},
"dependencies": {
"@nestjs/common": "^10.4.0",
"@nestjs/core": "^10.4.0",
"@nestjs/platform-express": "^10.4.0",
"drizzle-orm": "^0.31.0",
"mysql2": "^3.11.0",
"kafkajs": "^2.2.0",
"@elastic/elasticsearch": "^8.15.0",
"pino": "^9.4.0",
"prom-client": "^15.1.0",
"@opentelemetry/api": "^1.9.0",
"zod": "^3.23.0",
"uuid": "^10.0.0",
"reflect-metadata": "^0.2.2",
"rxjs": "^7.8.0",
"@opentelemetry/sdk-node": "^0.55.0",
"@opentelemetry/exporter-trace-otlp-http": "^0.55.0"
},
"devDependencies": {
"@nestjs/cli": "^10.4.0",
"@types/node": "^22.0.0",
"typescript": "^5.6.0",
"vitest": "^2.1.0"
}
}

View File

@@ -0,0 +1,7 @@
import { Module } from '@nestjs/common';
import { NotificationsModule } from './notifications/notifications.module.js';
@Module({
imports: [NotificationsModule],
})
export class AppModule {}

View File

@@ -0,0 +1,24 @@
import { drizzle } from 'drizzle-orm/mysql2';
import mysql from 'mysql2/promise';
import { env } from './env.js';
let pool: mysql.Pool | null = null;
export function getDb() {
if (!pool) {
pool = mysql.createPool({
uri: env.DATABASE_URL,
waitForConnections: true,
connectionLimit: 10,
queueLimit: 0,
});
}
return drizzle(pool);
}
export async function closeDb(): Promise<void> {
if (pool) {
await pool.end();
pool = null;
}
}

View File

@@ -0,0 +1,13 @@
import { Client } from '@elastic/elasticsearch';
import { env } from './env.js';
export const esClient = new Client({ node: env.ES_URL });
export async function checkEsConnection(): Promise<void> {
try {
await esClient.ping();
console.log('Elasticsearch connected');
} catch (err) {
console.error('Elasticsearch connection failed:', err);
}
}

View File

@@ -0,0 +1,27 @@
import { z } from 'zod';
const envSchema = z.object({
PORT: z.string().default('3007'),
DATABASE_URL: z.string().url(),
REDIS_URL: z.string().url().optional(),
JWT_SECRET: z.string(),
JWT_ISSUER: z.string().default('next-edu-cloud'),
KAFKA_BROKERS: z.string().default('localhost:9092'),
ES_URL: z.string().url().default('http://localhost:9200'),
OTEL_EXPORTER_OTLP_ENDPOINT: z.string().url().optional(),
LOG_LEVEL: z.enum(['fatal', 'error', 'warn', 'info', 'debug', 'trace']).default('info'),
NODE_ENV: z.enum(['development', 'production', 'test']).default('development'),
});
export type Env = z.infer<typeof envSchema>;
export function loadEnv(): Env {
const result = envSchema.safeParse(process.env);
if (!result.success) {
console.error('❌ Invalid environment variables:', result.error.flatten().fieldErrors);
throw new Error('Invalid environment configuration');
}
return result.data;
}
export const env = loadEnv();

31
services/msg/src/main.ts Normal file
View File

@@ -0,0 +1,31 @@
import 'reflect-metadata';
import { NestFactory } from '@nestjs/core';
import { AppModule } from './app.module.js';
import { GlobalErrorFilter } from './shared/errors/global-error.filter.js';
import { initTracer, shutdownTracer } from './shared/observability/tracer.js';
import { env } from './config/env.js';
import { logger } from './shared/observability/logger.js';
async function bootstrap(): Promise<void> {
initTracer();
const app = await NestFactory.create(AppModule, {
logger: ['log', 'error', 'warn'],
});
app.useGlobalFilters(new GlobalErrorFilter());
app.enableShutdownHooks();
await app.listen(env.PORT);
logger.info({ port: env.PORT }, 'Msg service started');
process.on('SIGTERM', async () => {
await app.close();
await shutdownTracer();
});
}
bootstrap().catch((err: unknown) => {
logger.error({ err }, 'Failed to start msg service');
process.exit(1);
});

View File

@@ -0,0 +1,32 @@
import { Body, Controller, Get, Param, Post, Query, Req } from '@nestjs/common';
import { NotificationsService } from './notifications.service.js';
import type { SendNotificationDto } from './notifications.service.js';
@Controller('notifications')
export class NotificationsController {
constructor(private readonly service: NotificationsService) {}
@Post()
async send(@Body() body: unknown) {
const result = await this.service.send(body as SendNotificationDto);
return { success: true, data: result };
}
@Get()
async list(@Req() req: { userId: string }, @Query('unread') unread: string) {
const result = await this.service.listByUser(req.userId, unread === 'true');
return { success: true, data: result };
}
@Post(':id/read')
async markAsRead(@Param('id') id: string) {
await this.service.markAsRead(id);
return { success: true };
}
@Get('search')
async search(@Req() req: { userId: string }, @Query('q') q: string) {
const result = await this.service.search(req.userId, q);
return { success: true, data: result };
}
}

View File

@@ -0,0 +1,9 @@
import { Module } from '@nestjs/common';
import { NotificationsController } from './notifications.controller.js';
import { NotificationsService } from './notifications.service.js';
@Module({
controllers: [NotificationsController],
providers: [NotificationsService],
})
export class NotificationsModule {}

View File

@@ -0,0 +1,24 @@
import { mysqlTable, varchar, char, timestamp, text, boolean, json } from 'drizzle-orm/mysql-core';
export const notifications = mysqlTable('msg_notifications', {
id: char('id', { length: 36 }).notNull().primaryKey(),
userId: char('user_id', { length: 36 }).notNull(),
type: varchar('type', { length: 50 }).notNull(),
title: varchar('title', { length: 200 }).notNull(),
content: text('content').notNull(),
channel: varchar('channel', { length: 20 }).notNull().default('in_app'),
isRead: boolean('is_read').notNull().default(false),
metadata: json('metadata'),
createdAt: timestamp('created_at').notNull().defaultNow(),
});
export const notificationPreferences = mysqlTable('msg_notification_preferences', {
userId: char('user_id', { length: 36 }).notNull().primaryKey(),
emailEnabled: boolean('email_enabled').notNull().default(true),
smsEnabled: boolean('sms_enabled').notNull().default(false),
pushEnabled: boolean('push_enabled').notNull().default(true),
inAppEnabled: boolean('in_app_enabled').notNull().default(true),
});
export type Notification = typeof notifications.$inferSelect;
export type NotificationPreference = typeof notificationPreferences.$inferSelect;

View File

@@ -0,0 +1,110 @@
import { Injectable } from '@nestjs/common';
import { getDb } from '../config/database.js';
import { notifications, notificationPreferences } from './notifications.schema.js';
import type { NotificationPreference } from './notifications.schema.js';
import { v4 as uuidv4 } from 'uuid';
import { eq, and } from 'drizzle-orm';
import { esClient } from '../config/elasticsearch.js';
import { logger } from '../shared/observability/logger.js';
export interface SendNotificationDto {
userId: string;
type: string;
title: string;
content: string;
channel?: string;
metadata?: Record<string, unknown>;
}
@Injectable()
export class NotificationsService {
async send(dto: SendNotificationDto) {
const id = uuidv4();
const db = getDb();
// 检查用户偏好
const [pref] = await db.select().from(notificationPreferences).where(eq(notificationPreferences.userId, dto.userId));
const channel = dto.channel || 'in_app';
if (pref && !this.isChannelEnabled(pref, channel)) {
logger.info({ userId: dto.userId, channel }, 'Notification skipped by preference');
return { skipped: true };
}
// 写入 DB
await db.insert(notifications).values({
id,
userId: dto.userId,
type: dto.type,
title: dto.title,
content: dto.content,
channel,
metadata: dto.metadata,
});
// 索引到 ES全文检索
try {
await esClient.index({
index: 'notifications',
id,
document: {
userId: dto.userId,
type: dto.type,
title: dto.title,
content: dto.content,
channel,
createdAt: new Date().toISOString(),
},
});
} catch (err) {
logger.error({ err }, 'Failed to index notification in ES');
}
// TODO: 触发 Push Gateway 推送P5 后期)
return { id, skipped: false };
}
async listByUser(userId: string, onlyUnread: boolean = false) {
const db = getDb();
const conditions = onlyUnread
? and(eq(notifications.userId, userId), eq(notifications.isRead, false))
: eq(notifications.userId, userId);
return db.select().from(notifications).where(conditions);
}
async markAsRead(id: string) {
const db = getDb();
await db.update(notifications).set({ isRead: true }).where(eq(notifications.id, id));
}
async search(userId: string, query: string) {
const result = await esClient.search({
index: 'notifications',
query: {
bool: {
must: [
{ term: { userId } },
{
multi_match: {
query,
fields: ['title', 'content'],
},
},
],
},
},
});
return result.hits.hits;
}
private isChannelEnabled(pref: NotificationPreference, channel: string): boolean {
switch (channel) {
case 'email': return pref.emailEnabled;
case 'sms': return pref.smsEnabled;
case 'push': return pref.pushEnabled;
case 'in_app': return pref.inAppEnabled;
default: return true;
}
}
}

View File

@@ -0,0 +1,96 @@
export type ErrorType =
| 'validation'
| 'not_found'
| 'permission_denied'
| 'conflict'
| 'business'
| 'database'
| 'internal';
export interface ErrorDetails {
[key: string]: unknown;
}
export abstract class ApplicationError extends Error {
abstract readonly type: ErrorType;
abstract readonly statusCode: number;
readonly code: string;
readonly details?: ErrorDetails;
// FIX #1: traceId 改为可写,以便 GlobalErrorFilter 注入请求级 traceId
traceId?: string;
constructor(message: string, code: string, details?: ErrorDetails) {
super(message);
this.name = this.constructor.name;
this.code = code;
this.details = details;
}
toJSON(): Record<string, unknown> {
return {
success: false,
error: {
code: this.code,
message: this.message,
details: this.details,
traceId: this.traceId,
},
};
}
}
export class ValidationError extends ApplicationError {
readonly type = 'validation' as const;
readonly statusCode = 400;
constructor(message: string, details?: ErrorDetails) {
super(message, 'MSG_VALIDATION_ERROR', details);
}
}
export class NotFoundError extends ApplicationError {
readonly type = 'not_found' as const;
readonly statusCode = 404;
constructor(resource: string, id: string) {
super(`${resource} not found: ${id}`, 'MSG_NOT_FOUND', { resource, id });
}
}
export class PermissionDeniedError extends ApplicationError {
readonly type = 'permission_denied' as const;
readonly statusCode = 403;
constructor(permission: string) {
super(`Permission denied: ${permission}`, 'MSG_PERMISSION_DENIED', { permission });
}
}
export class ConflictError extends ApplicationError {
readonly type = 'conflict' as const;
readonly statusCode = 409;
constructor(message: string, details?: ErrorDetails) {
super(message, 'MSG_CONFLICT', details);
}
}
export class BusinessError extends ApplicationError {
readonly type = 'business' as const;
readonly statusCode = 422;
constructor(message: string, details?: ErrorDetails) {
super(message, 'MSG_BUSINESS_ERROR', details);
}
}
export class DatabaseError extends ApplicationError {
readonly type = 'database' as const;
readonly statusCode = 500;
constructor(message: string, details?: ErrorDetails) {
super(message, 'MSG_DATABASE_ERROR', details);
}
}
export class InternalError extends ApplicationError {
readonly type = 'internal' as const;
readonly statusCode = 500;
constructor(message: string, details?: ErrorDetails) {
super(message, 'MSG_INTERNAL_ERROR', details);
}
}

View File

@@ -0,0 +1,77 @@
import { Catch, ExceptionFilter, ArgumentsHost, HttpException, Logger } from '@nestjs/common';
import { Request, Response } from 'express';
import { ZodError } from 'zod';
import { ApplicationError } from './application-error.js';
@Catch()
export class GlobalErrorFilter implements ExceptionFilter {
private readonly logger = new Logger(GlobalErrorFilter.name);
catch(exception: unknown, host: ArgumentsHost): void {
const ctx = host.switchToHttp();
const response = ctx.getResponse<Response>();
const request = ctx.getRequest<Request>();
const traceId = (request.headers['x-request-id'] as string | undefined) ?? 'unknown';
let statusCode = 500;
let body: Record<string, unknown>;
if (exception instanceof ApplicationError) {
exception.traceId = traceId;
statusCode = exception.statusCode;
body = exception.toJSON();
} else if (exception instanceof ZodError) {
// FIX #3: 捕获 Zod 解析错误,转换为结构化 ValidationError 响应
statusCode = 400;
body = {
success: false,
error: {
code: 'MSG_VALIDATION_ERROR',
message: 'Validation failed',
details: exception.flatten(),
traceId,
},
};
} else if (exception instanceof HttpException) {
statusCode = exception.getStatus();
const res = exception.getResponse();
const message = this.extractHttpMessage(res, exception);
body = {
success: false,
error: {
code: 'HTTP_ERROR',
message,
traceId,
},
};
} else {
this.logger.error(
`Unhandled exception: ${exception}`,
exception instanceof Error ? exception.stack : undefined,
);
body = {
success: false,
error: {
code: 'INTERNAL_ERROR',
message: 'An unexpected error occurred',
traceId,
},
};
}
response.status(statusCode).json(body);
}
private extractHttpMessage(res: string | object, exception: HttpException): string {
if (typeof res === 'string') {
return res;
}
if (res && typeof res === 'object' && 'message' in res) {
// 从 HttpException 响应体收窄类型NestJS 约定包含 message 字段)
const msg = (res as { message: unknown }).message;
return typeof msg === 'string' ? msg : exception.message;
}
return exception.message;
}
}

View File

@@ -0,0 +1,49 @@
import { Controller, Get, HttpException, HttpStatus } from '@nestjs/common';
import { DataSource } from 'typeorm';
const SERVICE_NAME = 'msg';
/**
* 健康检查端点。
*
* - GET /healthzliveness仅返回进程存活不检查依赖。
* - GET /readyzreadiness检查 DB 连接,失败返回 503。
*
* 不需要鉴权,必须在路由白名单中放行。本控制器内容在 iam / core-edu /
* content / msg / classes 五个 NestJS 服务中一致,仅 SERVICE_NAME 不同。
*/
@Controller()
export class HealthController {
constructor(private readonly dataSource: DataSource) {}
@Get('healthz')
liveness(): { status: string; service: string; timestamp: string } {
return {
status: 'ok',
service: SERVICE_NAME,
timestamp: new Date().toISOString(),
};
}
@Get('readyz')
async readiness(): Promise<{ status: string; service: string; timestamp: string }> {
try {
await this.dataSource.query('SELECT 1');
return {
status: 'ok',
service: SERVICE_NAME,
timestamp: new Date().toISOString(),
};
} catch (error) {
throw new HttpException(
{
status: 'error',
service: SERVICE_NAME,
timestamp: new Date().toISOString(),
error: error instanceof Error ? error.message : 'database unreachable',
},
HttpStatus.SERVICE_UNAVAILABLE,
);
}
}
}

View File

@@ -0,0 +1,23 @@
import { Module } from '@nestjs/common';
import { HealthController } from './health.controller';
/**
* 健康检查模块。
*
* 集成说明(不修改 app.module.ts仅在 README 注释说明):
*
* 在 `app.module.ts` 的 imports 数组中加入 `HealthModule`
*
* ```ts
* import { HealthModule } from './shared/health/health.module';
*
* @Module({ imports: [ ..., HealthModule ], ... })
* export class AppModule {}
* ```
*
* DataSource 由 `TypeOrmModule.forRoot(...)` 提供,本模块无需额外 provider。
*/
@Module({
controllers: [HealthController],
})
export class HealthModule {}

View File

@@ -0,0 +1,63 @@
import { Inject, Injectable, Logger, OnApplicationShutdown, OnModuleInit } from '@nestjs/common';
import { DataSource } from 'typeorm';
import type { Redis } from 'ioredis';
import type { Producer } from 'kafkajs';
const SERVICE_NAME = 'msg';
/**
* 优雅停机服务。
*
* 信号处理由 NestJS 在 `app.listen` 之前调用 `app.enableShutdownHooks()`
* 触发SIGTERM / SIGINTNestJS 会依次调用 OnApplicationShutdown 钩子。
* K8s 配置 `terminationGracePeriodSeconds=60` 给予足够时间清理。
*
* 关闭顺序Kafka producer → Redis → DataSource。
* 先停外部消息生产避免新事件,再关缓存,最后关 DB。
*
* 集成说明(不修改 app.module.ts仅在 README 注释说明):
* - 在 `app.module.ts` 的 providers 中加入 `LifecycleService`。
* - 在 `main.ts` 中 `app.listen` 之前调用 `app.enableShutdownHooks()`。
*
* 依赖注入 token 约定(需与各服务 provider 注册一致):
* - DataSource由 `TypeOrmModule.forRoot()` 提供。
* - 'REDIS_CLIENT':需在对应模块注册 `{ provide: 'REDIS_CLIENT', useFactory: ... }`。
* - 'KAFKA_PRODUCER':需在对应模块注册 `{ provide: 'KAFKA_PRODUCER', useFactory: ... }`。
*/
@Injectable()
export class LifecycleService implements OnModuleInit, OnApplicationShutdown {
private readonly logger = new Logger(LifecycleService.name);
constructor(
private readonly dataSource: DataSource,
@Inject('REDIS_CLIENT') private readonly redis: Redis,
@Inject('KAFKA_PRODUCER') private readonly kafkaProducer: Producer,
) {}
onModuleInit(): void {
this.logger.log(`service ${SERVICE_NAME} module initialized`);
}
async onApplicationShutdown(signal?: string): Promise<void> {
this.logger.log(
`service ${SERVICE_NAME} shutting down (signal=${signal ?? 'unknown'})`,
);
await this.safeDisconnect('kafka producer', () => this.kafkaProducer.disconnect());
await this.safeDisconnect('redis', () => this.redis.quit());
await this.safeDisconnect('datasource', () => this.dataSource.destroy());
this.logger.log(`service ${SERVICE_NAME} shutdown complete`);
}
private async safeDisconnect(name: string, fn: () => Promise<unknown>): Promise<void> {
try {
await fn();
this.logger.log(`${name} closed`);
} catch (error) {
this.logger.error(
`${name} close failed: ${error instanceof Error ? error.message : String(error)}`,
);
}
}
}

View File

@@ -0,0 +1,20 @@
import pino from 'pino';
import { env } from '../../config/env.js';
export const logger = pino({
level: env.LOG_LEVEL,
// 修复pino 默认字段选项为 `base`,而非 `defaultFields`
base: {
service: 'msg',
version: '0.1.0',
},
transport:
env.NODE_ENV === 'development'
? {
target: 'pino-pretty',
options: { colorize: true },
}
: undefined,
});
export type Logger = typeof logger;

View File

@@ -0,0 +1,24 @@
import promClient from 'prom-client';
const registry = new promClient.Registry();
// 修复:在本地 registry 上设置默认标签(原代码误用全局 register
registry.setDefaultLabels({ service: 'msg' });
registry.registerMetric(
new promClient.Counter({
name: 'msg_requests_total',
help: 'Total number of msg requests',
labelNames: ['method', 'endpoint', 'status'],
}),
);
registry.registerMetric(
new promClient.Histogram({
name: 'msg_request_duration_seconds',
help: 'Msg request duration in seconds',
labelNames: ['method', 'endpoint'],
buckets: [0.01, 0.05, 0.1, 0.3, 0.5, 1, 3, 5],
}),
);
export { registry as metricsRegistry };

View File

@@ -0,0 +1,25 @@
import { NodeSDK } from '@opentelemetry/sdk-node';
import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-http';
import { env } from '../../config/env.js';
let sdk: NodeSDK | null = null;
export function initTracer(): void {
if (!env.OTEL_EXPORTER_OTLP_ENDPOINT) return;
sdk = new NodeSDK({
serviceName: 'msg',
traceExporter: new OTLPTraceExporter({
url: `${env.OTEL_EXPORTER_OTLP_ENDPOINT}/v1/traces`,
}),
});
sdk.start();
console.log('Tracer initialized');
}
export async function shutdownTracer(): Promise<void> {
if (sdk) {
await sdk.shutdown();
}
}

View File

@@ -0,0 +1,15 @@
{
"extends": "../../tsconfig.base.json",
"compilerOptions": {
"module": "NodeNext",
"moduleResolution": "NodeNext",
"target": "ES2022",
"experimentalDecorators": true,
"emitDecoratorMetadata": true,
"outDir": "./dist",
"rootDir": "./src",
"types": ["node"]
},
"include": ["src/**/*"],
"exclude": ["node_modules", "dist", "test"]
}

View File

@@ -0,0 +1,13 @@
FROM golang:1.22-alpine AS builder
WORKDIR /app
COPY go.mod go.sum ./
RUN go mod download
COPY . .
RUN CGO_ENABLED=0 go build -o push-gateway .
FROM alpine:3.20
RUN apk --no-cache add ca-certificates
WORKDIR /app
COPY --from=builder /app/push-gateway .
EXPOSE 8081
CMD ["./push-gateway"]

View File

@@ -0,0 +1,42 @@
# Push Gateway 推送网关服务
> 版本0.1P5 骨架)
> 端口8081
## 职责
实时推送基础设施服务Go 实现),管理 WebSocket 长连接。
接收 Msg 服务的推送请求,维护用户在线连接池,将消息实时投递到浏览器/移动端。
## 技术栈
- Go 1.22 + Gin 1.10
- gorilla/websocket 1.5
- golang-jwt/jwt/v5JWT 鉴权)
- zap结构化日志骨架
## 开发
```bash
go mod tidy
go run main.go # :8081
```
## API
| 方法 | 路径 | 说明 |
|------|------|------|
| GET | /healthz | 健康检查 |
| GET | /ws?token=JWT | WebSocket 升级端点JWT 鉴权) |
| POST | /internal/push | 内部推送 APIMsg 服务调用) |
## 环境变量
| 变量 | 默认值 | 说明 |
|------|--------|------|
| PUSH_GATEWAY_PORT | 8081 | 服务端口 |
| JWT_SECRET | p1-dev-secret-change-in-production | JWT 密钥 |
## WebSocket 心跳
客户端发送 `ping` 文本帧,服务端回复 `pong`

View File

@@ -0,0 +1,11 @@
module github.com/edu-cloud/push-gateway
go 1.22
require (
github.com/gin-gonic/gin v1.10.0
github.com/golang-jwt/jwt/v5 v5.2.1
github.com/google/uuid v1.6.0
github.com/gorilla/websocket v1.5.3
go.uber.org/zap v1.27.0
)

View File

@@ -0,0 +1,22 @@
package config
import "os"
type Config struct {
Port string
JWTSecret string
}
func Load() *Config {
return &Config{
Port: getEnv("PUSH_GATEWAY_PORT", "8081"),
JWTSecret: getEnv("JWT_SECRET", "p1-dev-secret-change-in-production"),
}
}
func getEnv(key, fallback string) string {
if v := os.Getenv(key); v != "" {
return v
}
return fallback
}

View File

@@ -0,0 +1,59 @@
package hub
import (
"sync"
"github.com/gorilla/websocket"
)
// Hub 管理 WebSocket 客户端连接
type Hub struct {
mu sync.RWMutex
clients map[string]map[*websocket.Conn]bool // userID -> connections
}
func NewHub() *Hub {
return &Hub{
clients: make(map[string]map[*websocket.Conn]bool),
}
}
func (h *Hub) Register(userID string, conn *websocket.Conn) {
h.mu.Lock()
defer h.mu.Unlock()
if h.clients[userID] == nil {
h.clients[userID] = make(map[*websocket.Conn]bool)
}
h.clients[userID][conn] = true
}
func (h *Hub) Unregister(userID string, conn *websocket.Conn) {
h.mu.Lock()
defer h.mu.Unlock()
if conns, ok := h.clients[userID]; ok {
delete(conns, conn)
if len(conns) == 0 {
delete(h.clients, userID)
}
}
}
// SendToUser 向指定用户的所有连接推送消息
func (h *Hub) SendToUser(userID string, message []byte) error {
h.mu.RLock()
defer h.mu.RUnlock()
conns, ok := h.clients[userID]
if !ok {
return nil
}
for conn := range conns {
if err := conn.WriteMessage(websocket.TextMessage, message); err != nil {
return err
}
}
return nil
}

View File

@@ -0,0 +1,99 @@
package ws
import (
"net/http"
"strings"
"github.com/edu-cloud/push-gateway/internal/hub"
"github.com/gin-gonic/gin"
"github.com/golang-jwt/jwt/v5"
"github.com/gorilla/websocket"
)
var upgrader = websocket.Upgrader{
CheckOrigin: func(r *http.Request) bool {
return true // P5 骨架,生产环境需校验 origin
},
}
type Handler struct {
hub *hub.Hub
jwtSecret string
}
func NewHandler(h *hub.Hub, jwtSecret string) *Handler {
return &Handler{hub: h, jwtSecret: jwtSecret}
}
func (h *Handler) HandleWebSocket(c *gin.Context) {
// 从 query 参数获取 tokenWebSocket 无法设置 Authorization 头)
tokenStr := c.Query("token")
if tokenStr == "" {
c.JSON(http.StatusUnauthorized, gin.H{"error": "missing token"})
return
}
token, err := jwt.Parse(tokenStr, func(t *jwt.Token) (interface{}, error) {
if _, ok := t.Method.(*jwt.SigningMethodHMAC); !ok {
return nil, jwt.ErrSignatureInvalid
}
return []byte(h.jwtSecret), nil
})
if err != nil || !token.Valid {
c.JSON(http.StatusUnauthorized, gin.H{"error": "invalid token"})
return
}
claims, ok := token.Claims.(jwt.MapClaims)
if !ok {
c.JSON(http.StatusUnauthorized, gin.H{"error": "invalid claims"})
return
}
userID, ok := claims["sub"].(string)
if !ok {
c.JSON(http.StatusUnauthorized, gin.H{"error": "missing user id"})
return
}
conn, err := upgrader.Upgrade(c.Writer, c.Request, nil)
if err != nil {
return
}
defer conn.Close()
h.hub.Register(userID, conn)
defer h.hub.Unregister(userID, conn)
// 读取循环(保持连接,处理心跳)
for {
_, msg, err := conn.ReadMessage()
if err != nil {
break
}
// 处理心跳 ping
if strings.ToLower(string(msg)) == "ping" {
conn.WriteMessage(websocket.TextMessage, []byte("pong"))
}
}
}
// PushHandler 接收来自 Msg 服务的推送请求
func (h *Handler) PushHandler(c *gin.Context) {
var req struct {
UserID string `json:"user_id"`
Message string `json:"message"`
}
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"success": false, "error": gin.H{"code": "INVALID_REQUEST", "message": err.Error()}})
return
}
if err := h.hub.SendToUser(req.UserID, []byte(req.Message)); err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"success": false, "error": gin.H{"code": "PUSH_FAILED", "message": err.Error()}})
return
}
c.JSON(http.StatusOK, gin.H{"success": true})
}

View File

@@ -0,0 +1,63 @@
package main
import (
"context"
"log"
"net/http"
"os"
"os/signal"
"syscall"
"time"
"github.com/edu-cloud/push-gateway/internal/config"
"github.com/edu-cloud/push-gateway/internal/hub"
"github.com/edu-cloud/push-gateway/internal/ws"
"github.com/gin-gonic/gin"
)
func main() {
cfg := config.Load()
gin.SetMode(gin.ReleaseMode)
h := hub.NewHub()
wsHandler := ws.NewHandler(h, cfg.JWTSecret)
r := gin.New()
r.Use(gin.Recovery())
r.GET("/healthz", func(c *gin.Context) {
c.JSON(200, gin.H{"status": "ok", "service": "push-gateway"})
})
// WebSocket 升级端点
r.GET("/ws", wsHandler.HandleWebSocket)
// 内部推送 APIMsg 服务调用)
api := r.Group("/internal")
api.POST("/push", wsHandler.PushHandler)
srv := &http.Server{
Addr: ":" + cfg.Port,
Handler: r,
ReadTimeout: 10 * time.Second,
WriteTimeout: 60 * time.Second, // WebSocket 长连接
}
go func() {
log.Printf("Push Gateway listening on :%s", cfg.Port)
if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
log.Fatalf("listen: %s\n", err)
}
}()
quit := make(chan os.Signal, 1)
signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
<-quit
log.Println("Shutting down Push Gateway...")
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
if err := srv.Shutdown(ctx); err != nil {
log.Fatal("Server forced to shutdown:", err)
}
}