diff --git a/docker/tinode/config.template b/docker/tinode/config.template index 86fa056..17cbb0f 100644 --- a/docker/tinode/config.template +++ b/docker/tinode/config.template @@ -271,11 +271,7 @@ "cluster_config": { "self": "", - "nodes": [ - {"name": "tinode-0", "addr": "tinode-0:12000"}, - {"name": "tinode-1", "addr": "tinode-1:12001"}, - {"name": "tinode-2", "addr": "tinode-2:12002"} - ], + "nodes": $CLUSTER_NODES_JSON, "failover": { "enabled": true, "heartbeat": 100, diff --git a/docker/tinode/entrypoint.sh b/docker/tinode/entrypoint.sh index 661f7cd..c2d98a0 100644 --- a/docker/tinode/entrypoint.sh +++ b/docker/tinode/entrypoint.sh @@ -13,6 +13,9 @@ else CONFIG=working.config : "${REDIS_KEY_PREFIX:=im}" + if [ -z "$CLUSTER_NODES_JSON" ] ; then + CLUSTER_NODES_JSON='[{"name":"tinode-0","addr":"tinode-0:12000"},{"name":"tinode-1","addr":"tinode-1:12001"},{"name":"tinode-2","addr":"tinode-2:12002"}]' + fi # Remove the old config. rm -f working.config diff --git a/docs/IM-原生Cluster与RedisFanout职责边界.md b/docs/IM-原生Cluster与RedisFanout职责边界.md new file mode 100644 index 0000000..598b4b5 --- /dev/null +++ b/docs/IM-原生Cluster与RedisFanout职责边界.md @@ -0,0 +1,256 @@ +# IM 原生 Cluster 与 Redis Fanout 职责边界 + +## 1. 背景 + +IM Server 当前同时存在两套跨 Pod 能力： + +- Tinode 原生 Cluster：以 Topic master 为中心，负责把 Topic 请求路由到唯一 master 节点，由 master 统一处理 Topic 状态机。 +- Redis Fanout：以 Redis Pub/Sub 为跨 Pod 实时总线，负责把服务端生成的实时事件广播到其他 Pod，再由各 Pod 投递给本机 WebSocket session。 + +测试环境目前是多 Pod standalone 部署，日志表现为 `Cluster: running as a standalone server.`，因此当前实际跨 Pod 主链路是 Redis Fanout。目标架构需要启用原生 Cluster，把 Topic 热路径、强顺序状态机和实时通话状态迁回 Topic master；Redis Fanout 保留为大数据量、缓存、补偿和用户级控制能力。 + +本文定义两套机制的职责边界，避免后续同时启用后出现重复投递、重复状态变更或状态源不一致。 + +## 2. 总原则 + +1. 同一类 Topic 事件只能有一个实时主链路。 +2. 原生 Cluster 启用时，Topic 状态机相关事件优先由 Cluster 处理。 +3. Redis Fanout 不作为 Topic master 的替代品；它是跨 Pod 投递、通知和 session control 总线。 +4. 关键业务状态必须以 MySQL、Redis 原子状态或 Kafka 顺序消费作为权威，WebSocket fanout 只负责实时通知。 +5. 可丢的临时状态可以只走 fanout；不可丢的状态必须有历史补拉或条件更新兜底。 +6. 能利用原生 Cluster 内存态、高性能、强一致和强顺序的 Topic 热路径，应优先交给原生 Cluster。 +7. 数据量大、需要广播、缓存、聚合、补偿或跨业务模块共享的数据，应优先交给 Redis/MySQL/Kafka。 + +## 3. 分工原则 + +两套机制不是互斥关系，而是按数据特征分工： + +```text +强状态、强顺序、小范围热路径 -> 原生 Cluster +大数据量、跨节点广播、缓存、补偿 -> Redis / MySQL / Kafka +``` + +### 3.1 优先使用原生 Cluster 的场景 + +以下场景应尽量使用原生 Cluster，发挥 Topic master 的内存态和串行状态机能力： + +- 单个 Topic 内的订阅、离开、权限变更和在线 session 状态。 +- 单聊/群聊 Topic 内的 `pub`、`read`、`recv`、`get/set meta`。 +- 通话实时状态机：`ringing`、`accept`、`hang-up`、`offer/answer/ice-candidate`。 +- 对顺序敏感的 Topic 内临时事件，例如正在输入、同 Topic 内 presence 派发。 +- 需要立即反馈客户端、且依赖 Topic 当前内存状态的操作。 + +这些能力的共同特点是： + +- 数据范围集中在单个 Topic。 +- 需要强顺序或强一致。 +- 更适合由唯一 Topic master 串行处理。 +- 不适合广播给所有 Pod 后各自判断。 + +### 3.2 优先使用 Redis / MySQL / Kafka 的场景 + +以下场景应优先使用 Redis、MySQL 或 Kafka，不强行塞进原生 Cluster： + +- 全局用户在线状态查询、好友列表在线状态展示。 +- 群成员在线节点集合、大群在线成员筛选。 +- 大群 fanout、热点消息缓存、未读聚合、离线补偿。 +- 用户级 session control：踢端、全端下线、同平台挤旧端、密码重置清退。 +- 限流、短状态、幂等、分布式去重。 +- 历史消息、通话记录、审计、后台统计。 +- Push、文件安全、后台管理命令等异步任务。 + +这些能力的共同特点是： + +- 数据量大或跨 Topic。 +- 需要被 HTTP、Kafka、后台任务等多入口共享。 +- 可以最终一致，或可以通过历史补拉/条件更新补偿。 +- 不适合放在某个 Topic master 的内存里长期维护。 + +## 4. 目标启用模式 + +目标模式为： + +```text +原生 Cluster Topic master + Redis 大数据量支撑 + MySQL/Kafka 权威状态 +``` + +目标分工： + +- 原生 Cluster 负责单个 Topic 内的实时请求路由、内存状态机、强顺序处理和低延迟下发。 +- Redis 负责全局在线状态、大群在线成员集合、热点缓存、大数据量 fanout、限流、幂等和 session control。 +- MySQL 负责最终历史、通话记录、关系状态和可补偿查询。 +- Kafka 负责后台异步任务、Push、审计、文件安全和跨服务事件。 + +启用原生 Cluster 后的核心原则： + +```text +Topic 内强状态、强顺序、小范围热路径，必须优先走原生 Cluster +跨 Topic、大数据量、可最终一致或需补偿的数据，继续走 Redis/MySQL/Kafka +``` + +当前 standalone + Redis Fanout 是过渡状态。切换到原生 Cluster 前，需要先完成部署形态、职责收敛、重复链路关闭和压测验收。 + +## 5. 职责边界表 + +| 能力 | 当前 standalone + Redis Fanout 过渡模式 | 目标原生 Cluster 模式 | 说明 | +| --- | --- | --- | --- | +| Topic master 路由 | 不启用 | Cluster 负责 | 原生 Cluster 通过一致性哈希确定 Topic master。 | +| `{sub}` 订阅 | 本 Pod Hub/Topic 处理，跨 Pod 投递依赖 Redis | Cluster `ProxyReqJoin` | 启 Cluster 后不要再用 Redis 模拟订阅路由。 | +| `{leave}` 离开 | 本 Pod 处理 | Cluster `ProxyReqLeave` | 与订阅同源。 | +| `{pub}` 消息写入 | 本 Pod 处理，生成 data 后 Redis fanout | Cluster `ProxyReqBroadcast` 到 Topic master | 启 Cluster 后 Topic data 不应再额外 Redis topic fanout。 | +| `{get/set}` meta | 本 Pod 处理 | Cluster `ProxyReqMeta` | 强依赖 Topic 状态时应由 master 处理。 | +| `read/recv` 回执 | Redis topic/user fanout 下发结果 | Cluster 路由 note 到 master | 启 Cluster 后回执状态由 master 生成，避免双份 info。 | +| 通话 `ringing/accept/hang-up` | Redis `call_event` fanout + DB 条件更新兜底 | Cluster `ProxyReqCall` | 启 Cluster 后 Redis `call_event` 必须跳过。 | +| `offer/answer/ice-candidate` | 当前可由现有通话状态机处理；声网模式下不走 SDP/ICE | Cluster 由 Topic master 转发 | 首版声网主要关注 accept/hang-up。 | +| Topic 内 data/info/pres 下发 | Redis realtime fanout | Cluster proxy response | 两者不能同时作为主链路。 | +| 用户在线状态查询 | Redis presence | 可继续 Redis presence | 即使启 Cluster，也建议 Redis 保持全局在线查询缓存。 | +| 群在线成员缓存 | Redis topic member presence | 可继续 Redis 缓存 | 用于大群本节点在线成员查找和展示。 | +| 踢端/全端下线 | Redis session control | Redis session control | 用户级控制不属于单个 Topic，建议继续 Redis。 | +| 密码重置清退 | Redis session control | Redis session control | 与 Cluster 不冲突。 | +| 同平台新登录挤旧端 | 本 Pod + Redis platform evict | 本 Pod + Redis platform evict | 继续用 Redis 保证跨 Pod 生效。 | +| 好友/群业务事件 | 本地投递 + Redis user/topic fanout | 可继续 Redis user fanout | 业务事件通常不是 Topic 状态机请求。 | +| 系统消息通知 | 本地投递 + Redis fanout | 视消息写入入口决定 | 如果写入会话 Topic，应避免和 Cluster topic 下发重复。 | +| 群消息 seq/热消息缓存 | Redis / MySQL | 可继续 Redis / MySQL | 大数据量缓存和历史权威不依赖原生 Cluster。 | +| Kafka 后台事件 | Kafka 负责异步持久化、Push、审计 | Kafka 继续负责 | Kafka 不做实时 WS fanout 主链路。 | + +## 6. Redis Fanout 主链路范围 + +在当前 standalone 过渡模式下，Redis Fanout 是跨 Pod 实时投递主链路。启用原生 Cluster 后，Redis Fanout 需要收敛为配套能力，负责： + +1. `message/user`：用户维度的系统事件、业务事件、me presence 下发。 +2. `session_control/user|device|platform`：踢端、清退、同平台挤旧端。 +3. presence 缓存：用户在线状态、节点 session 信息、群成员在线节点信息。 +4. 大群在线成员筛选、热点消息缓存、未读聚合和历史补偿辅助。 +5. 限流、短状态、幂等、分布式去重。 + +仅在 standalone 过渡模式下，Redis Fanout 额外负责： + +1. `message/topic`：Topic 维度的 `data`、`info`、`pres` 跨 Pod下发。 +2. `call_event/topic`：通话 `ringing`、`accept`、`hang-up` 进入 Topic 状态机。 + +这些 Topic 状态机相关 fanout 在原生 Cluster 启用后必须关闭或跳过。 + +Redis Fanout 不负责： + +- 作为消息历史权威。 +- 作为通话最终状态权威。 +- 保证同一 Topic 的全局严格顺序。 +- 替代 MySQL/Kafka 的审计、补偿和持久化。 + +## 7. 原生 Cluster 主链路范围 + +启用原生 Cluster 后，它应负责： + +1. Topic owner/master 分配。 +2. Topic 请求路由：`sub`、`leave`、`pub`、`get/set`、`note`。 +3. Topic 内状态机：订阅、权限、回执、通话、presence 派发。 +4. Topic master 到 proxy session 的响应和广播。 +5. 节点故障后的 proxy session 清理与订阅失效通知。 + +原生 Cluster 不建议负责： + +- 用户级 session control。 +- 全局在线状态查询缓存。 +- 业务后台事件可靠投递。 +- Kafka 异步消费、Push、审计。 +- Redis 热消息缓存、限流、短状态。 +- 大群全量成员 fanout、全局未读聚合、历史数据扫描等大数据量任务。 + +## 8. 双开时必须避免的重复链路 + +如果 `globals.cluster != nil`，以下 Redis Fanout 应跳过或降级： + +| Redis 入口 | Cluster 启用时处理 | +| --- | --- | +| `publishRealtimeTopicMessage` | 跳过。Topic data/info/pres 由 Cluster proxy response 下发。 | +| `publishRealtimeCallEvent` | 跳过。通话 note 由 `ProxyReqCall` 路由到 Topic master。 | +| 群消息在线投递 fanout | 需要专项确认；如果同一 data 已由 Cluster 下发，不得再广播。 | +| 会话 Topic 系统消息 fanout | 如果消息进入 Topic 状态机，应避免再次 topic fanout。 | + +以下 Redis Fanout 可继续保留： + +| Redis 入口 | 保留原因 | +| --- | --- | +| `publishRealtimeEvictUser` | 用户级清退，不归属单一 Topic。 | +| `publishRealtimeEvictDevice` | 设备级清退，不归属单一 Topic。 | +| `publishRealtimeEvictUserPlatform` | 同平台挤旧端，需要跨 Pod 控制 session。 | +| 用户 presence 缓存 | 全局在线查询和好友列表展示仍需要。 | +| 业务 user event | 多数业务事件不是 Topic 状态机请求。 | + +## 9. 状态权威与顺序规则 + +### 9.1 消息顺序 + +- 权威顺序来自 `topic + seq`。 +- Redis Fanout 到达乱序时，客户端必须按 seq 去重和重排。 +- 服务端历史读取按 seq 返回，作为重连补偿来源。 + +### 9.2 通话状态 + +通话状态必须使用条件更新保证单向流转： + +```text +started -> accepted -> finished +started -> missed +started -> declined +accepted -> disconnected +``` + +规则： + +- `accept` 只能从 `started` 写入 `accepted`。 +- `missed` 只能从 `started` 且 `accepted_at IS NULL` 写入。 +- `finished` 只能结束未结束通话。 +- 已 `accepted` 的通话不得被 `missed` 或 `declined` 覆盖。 +- WebSocket 下发失败不能改变最终状态，客户端可通过通话记录或会话消息补偿。 + +### 9.3 Session Control + +session control 以 Redis 短状态和 token/auth_version 为兜底： + +- Redis 实时踢端用于立即关闭在线 WS。 +- token/auth_version 用于防止旧 token 重新登录。 +- stale session-control fanout 必须丢弃，避免旧事件踢掉新 session。 + +## 10. 启用原生 Cluster 的前置条件 + +启用原生 Cluster 前必须满足： + +1. K8s 部署从普通 Deployment 改为适合稳定网络身份的形态，例如 StatefulSet 或稳定 Pod DNS。 +2. 每个 Pod 配置唯一且稳定的 `cluster_self`。 +3. 配置完整 `cluster_config.nodes`，包括节点名和可达地址。 +4. 开放并验证 gRPC/cluster 节点间通信。 +5. 明确 Redis Fanout 哪些保留、哪些关闭。 +6. 压测 `sub/pub/read/recv/call/group` 跨节点链路。 +7. 验证节点重启、扩缩容、Topic rehash、proxy subscription 失效和客户端重订阅。 +8. 验证不会重复投递 data/info/pres/call event。 + +## 11. 观测与验收 + +必须保留以下观测指标： + +- Redis fanout publish/receive/deliver 计数。 +- Redis fanout `skip_native_cluster` 计数。 +- Redis fanout `no_local_topic`、`no_local_session`、`hub_full` 计数。 +- 通话状态转换计数：started、accepted、finished、missed、declined。 +- DB 条件更新失败计数，用于识别迟到事件和乱序事件。 +- Cluster 模式下 proxy queue、TopicMaster RPC 失败、cluster unreachable、rehash 次数。 + +验收用例至少覆盖： + +1. A、B 连接到不同 Pod，P2P 消息实时到达。 +2. 接收方未订阅当前会话，仅在线 me，能收到未读提醒或补偿。 +3. 通话 accept/hang-up 跨 Pod 到达，且不会被 missed 覆盖。 +4. 多端登录同一账号，所有端收到业务事件。 +5. 密码重置或后台踢端能清退所有 Pod 上的 session。 +6. 大群在线成员分布在多 Pod，消息只投递在线成员，离线靠 seq 补拉。 +7. Redis 短暂不可用时，关键状态不写坏，客户端可补偿。 + +## 12. 当前结论 + +当前阶段职责划分如下： + +- 当前测试环境仍是 standalone + Redis Fanout，这是过渡状态。 +- 目标是启用原生 Cluster，让 Topic master 承担内存高性能、强一致、强顺序的实时状态机。 +- Redis 不退出架构，而是转为大数据量、缓存、补偿、全局 presence、session control 和异步支撑能力。 +- 启用原生 Cluster 时，Topic 状态机相关的 Redis topic/call fanout 必须关闭或降级，避免 data/info/pres/call event 双路投递。 diff --git a/docs/IM-原生Cluster运维部署说明.md b/docs/IM-原生Cluster运维部署说明.md new file mode 100644 index 0000000..adfc27d --- /dev/null +++ b/docs/IM-原生Cluster运维部署说明.md @@ -0,0 +1,430 @@ +# IM 原生 Cluster 运维部署说明 + +## 1. 目标 + +本文给运维说明 IM Server 如何部署才能真正启用 Tinode 原生 Cluster。 + +目标效果： + +- 多个 IM Pod 组成同一个 Tinode Cluster，不再是 standalone。 +- 每个 Pod 拥有稳定、唯一的节点名。 +- 节点之间通过 `cluster_config.nodes.addr` 建立内部 TCP RPC 通信。 +- Topic 请求可以路由到唯一 Topic master，由 master 处理 Topic 内状态机、顺序和实时下发。 + +如果部署后日志仍出现： + +```text +Cluster: running as a standalone server. +``` + +说明原生 Cluster 没有启用，跨 Pod Topic 状态机不会生效。 + +## 2. 原生 Cluster 启用条件 + +必须同时满足以下条件： + +1. `cluster_self` 非空。 +2. `cluster_self` 的值必须等于 `cluster_config.nodes` 中某个节点的 `name`。 +3. `cluster_config.nodes` 必须包含所有 IM 节点，包括当前节点。 +4. 每个节点的 `addr` 必须是其他 Pod 可访问的内部地址。 +5. 每个节点必须监听自己的 `addr` 端口。 +6. 节点之间网络策略、安全组、Service 必须放通 cluster 内部端口。 +7. 至少 2 个节点才算有效 Cluster，生产建议 3 个节点起。 + +当前代码启动参数由 `docker/tinode/entrypoint.sh` 传入： + +```bash +./tinode --config=${CONFIG} --static_data=$STATIC_DIR --cluster_self=$CLUSTER_SELF --pprof_url=$PPROF_URL +``` + +所以运维侧必须确保每个 Pod 的环境变量 `CLUSTER_SELF` 有值，并且值稳定。 + +## 3. 推荐部署形态 + +推荐使用 Kubernetes `StatefulSet`，不要用普通 `Deployment` 直接滚动多个副本。 + +原因： + +- 原生 Cluster 依赖稳定节点名，例如 `im-server-0`、`im-server-1`、`im-server-2`。 +- 原生 Cluster 依赖稳定 DNS，例如 `im-server-0.im-server-cluster.default.svc.cluster.local`。 +- 普通 Deployment 的 Pod 名会变化，不适合作为 `cluster_self`。 +- StatefulSet 配合 Headless Service 可以给每个 Pod 提供固定网络身份。 + +推荐结构： + +```text +Headless Service: im-server-cluster +StatefulSet: im-server +Pod DNS: + im-server-0.im-server-cluster..svc.cluster.local + im-server-1.im-server-cluster..svc.cluster.local + im-server-2.im-server-cluster..svc.cluster.local +``` + +## 4. 端口规划 + +至少需要区分三类端口： + +| 端口 | 用途 | 是否对公网暴露 | 说明 | +| --- | --- | --- | --- | +| `6060` | HTTP/WebSocket/API | 通过 Ingress/SLB 暴露 | 客户端连接入口。 | +| `16060` | Tinode gRPC 客户端端口 | 通常不对公网暴露 | 管理或内部 gRPC 客户端使用。 | +| `12000` | 原生 Cluster 内部 RPC | 不对公网暴露，只在 Pod 间放通 | `cluster_config.nodes.addr` 使用。 | + +注意：`cluster_config.nodes.addr` 使用的是原生 Cluster 内部 TCP RPC 地址，不是 `grpc_listen` 的 `16060`。 + +建议统一所有 Pod 的 cluster 监听端口为 `12000`，通过 Pod DNS 区分节点： + +```json +{ + "name": "im-server-0", + "addr": "im-server-0.im-server-cluster.default.svc.cluster.local:12000" +} +``` + +不要沿用 docker compose 示例里的 `tinode-0:12000`、`tinode-1:12001`、`tinode-2:12002`，除非是在同一主机或 docker compose 网络里部署。 + +## 5. 配置文件要求 + +`server/tinode.conf` 默认配置是： + +```json +"cluster_config": { + "self": "", + "nodes": [], + "failover": { + "enabled": false + } +} +``` + +这个配置不会启用 Cluster。 + +生产或测试环境需要提供完整配置。建议 3 节点配置如下： + +```json +"cluster_config": { + "self": "", + "nodes": [ + { + "name": "im-server-0", + "addr": "im-server-0.im-server-cluster.default.svc.cluster.local:12000" + }, + { + "name": "im-server-1", + "addr": "im-server-1.im-server-cluster.default.svc.cluster.local:12000" + }, + { + "name": "im-server-2", + "addr": "im-server-2.im-server-cluster.default.svc.cluster.local:12000" + } + ], + "failover": { + "enabled": true, + "heartbeat": 100, + "vote_after": 8, + "node_fail_after": 16 + } +} +``` + +说明： + +- `self` 可以保持为空，由启动参数 `--cluster_self=$CLUSTER_SELF` 覆盖。 +- `CLUSTER_SELF` 必须分别为 `im-server-0`、`im-server-1`、`im-server-2`。 +- `nodes` 必须所有 Pod 完全一致。 +- `addr` 必须能被其他 Pod 访问。 +- 3 节点建议开启 failover；2 节点不建议开启 failover。 + +## 6. Kubernetes 示例 + +以下示例只展示 Cluster 相关关键配置，实际部署时需要合并现有镜像、环境变量、Secret、资源限制、探针、Ingress 和业务配置。 + +### 6.1 Headless Service + +```yaml +apiVersion: v1 +kind: Service +metadata: + name: im-server-cluster + namespace: default +spec: + clusterIP: None + selector: + app: im-server + ports: + - name: cluster + port: 12000 + targetPort: 12000 + - name: http + port: 6060 + targetPort: 6060 + - name: grpc + port: 16060 + targetPort: 16060 +``` + +### 6.2 客户端入口 Service + +客户端入口可以继续使用普通 Service 或 Ingress，负载到所有 Pod： + +```yaml +apiVersion: v1 +kind: Service +metadata: + name: im-server + namespace: default +spec: + selector: + app: im-server + ports: + - name: http + port: 6060 + targetPort: 6060 +``` + +### 6.3 StatefulSet + +```yaml +apiVersion: apps/v1 +kind: StatefulSet +metadata: + name: im-server + namespace: default +spec: + serviceName: im-server-cluster + replicas: 3 + selector: + matchLabels: + app: im-server + template: + metadata: + labels: + app: im-server + spec: + containers: + - name: im-server + image: + ports: + - name: http + containerPort: 6060 + - name: grpc + containerPort: 16060 + - name: cluster + containerPort: 12000 + env: + - name: CLUSTER_SELF + valueFrom: + fieldRef: + fieldPath: metadata.name + - name: EXT_CONFIG + value: /etc/tinode/tinode.conf + volumeMounts: + - name: tinode-config + mountPath: /etc/tinode + readOnly: true + volumes: + - name: tinode-config + configMap: + name: im-server-tinode-conf +``` + +重点： + +- `serviceName` 必须指向 Headless Service。 +- `CLUSTER_SELF` 直接取 Pod 名，StatefulSet 下会稳定为 `im-server-0`、`im-server-1`、`im-server-2`。 +- ConfigMap 中的 `cluster_config.nodes.name` 必须和 Pod 名一致。 +- ConfigMap 中的 `cluster_config.nodes.addr` 建议使用 StatefulSet Pod DNS。 + +## 7. NetworkPolicy / 安全组要求 + +如果集群启用了 NetworkPolicy，需要允许 IM Pod 之间访问 cluster 端口： + +```yaml +apiVersion: networking.k8s.io/v1 +kind: NetworkPolicy +metadata: + name: im-server-cluster-allow + namespace: default +spec: + podSelector: + matchLabels: + app: im-server + policyTypes: + - Ingress + - Egress + ingress: + - from: + - podSelector: + matchLabels: + app: im-server + ports: + - protocol: TCP + port: 12000 + egress: + - to: + - podSelector: + matchLabels: + app: im-server + ports: + - protocol: TCP + port: 12000 +``` + +如果没有 NetworkPolicy，可以跳过此项，但仍要确认云厂商安全组、服务网格策略没有拦截 Pod 间 TCP `12000`。 + +## 8. 启动顺序 + +建议： + +1. 先部署 ConfigMap、Headless Service、普通入口 Service。 +2. 再部署 StatefulSet。 +3. 第一个 Pod 可以执行数据库初始化和升级。 +4. 其他 Pod 设置 `NO_DB_INIT=true`，避免多 Pod 同时初始化数据库。 +5. 等所有 Pod Ready 后再接入客户端流量。 + +如果当前镜像沿用 `docker/tinode/entrypoint.sh` 的逻辑： + +- 第一个 Pod 可以保留 `NO_DB_INIT=false`。 +- 其他 Pod 建议设置 `NO_DB_INIT=true`。 +- 如果所有 Pod 同时启动，要确认数据库初始化脚本具备幂等能力，否则建议改成 init Job 单独执行。 + +## 9. 发布步骤建议 + +### 9.1 灰度前检查 + +执行： + +```bash +kubectl -n default get pod -l app=im-server -o wide +kubectl -n default get svc im-server-cluster +kubectl -n default exec im-server-0 -- sh -c 'echo $CLUSTER_SELF' +kubectl -n default exec im-server-1 -- sh -c 'echo $CLUSTER_SELF' +kubectl -n default exec im-server-2 -- sh -c 'echo $CLUSTER_SELF' +``` + +期望： + +```text +im-server-0 +im-server-1 +im-server-2 +``` + +验证节点 DNS： + +```bash +kubectl -n default exec im-server-0 -- sh -c 'nc -vz im-server-1.im-server-cluster.default.svc.cluster.local 12000' +kubectl -n default exec im-server-0 -- sh -c 'nc -vz im-server-2.im-server-cluster.default.svc.cluster.local 12000' +``` + +### 9.2 启动后日志检查 + +查看日志： + +```bash +kubectl -n default logs im-server-0 | grep -i cluster +kubectl -n default logs im-server-1 | grep -i cluster +kubectl -n default logs im-server-2 | grep -i cluster +``` + +不能出现： + +```text +Cluster: running as a standalone server. +``` + +如果出现该日志，优先检查： + +- `CLUSTER_SELF` 是否为空。 +- `CLUSTER_SELF` 是否不在 `cluster_config.nodes.name` 中。 +- 配置文件是否确实被容器读取。 +- `EXT_CONFIG` 是否指向了正确文件。 +- `cluster_config.nodes` 是否为空。 + +如果出现节点连接失败，优先检查： + +- `cluster_config.nodes.addr` DNS 是否能解析。 +- 端口 `12000` 是否监听。 +- Pod 间 NetworkPolicy 是否放通。 +- Service `targetPort` 是否正确。 + +### 9.3 功能验收 + +至少验证： + +1. A、B 两个账号分别连到不同 Pod，P2P 消息实时互达。 +2. A、B 分别连到不同 Pod，语音通话 `ringing`、`accept`、`hang-up` 状态实时互达。 +3. 同一个账号多端连接不同 Pod，`recv/read` 回执正常。 +4. 群聊成员分布在不同 Pod，群消息实时下发正常。 +5. 重启任意一个 Pod 后，客户端能重连，Topic 可重新订阅。 +6. 日志无持续 cluster unreachable、proxy timeout、dial timeout。 + +## 10. 与 Redis Fanout 的关系 + +启用原生 Cluster 后，Redis 仍然保留，但职责要收敛： + +- 保留：全局 presence、session control、踢端、同平台挤旧端、大群在线缓存、热点缓存、限流、幂等、补偿。 +- 收敛或关闭：Topic data/info/pres 的 Redis topic fanout。 +- 收敛或关闭：通话 `call_event/topic` 的 Redis fanout。 + +否则会出现同一个 Topic 事件既由 Cluster 下发、又由 Redis Fanout 下发，导致重复 data/info/pres/call event。 + +对应职责边界参考： + +- `docs/IM-原生Cluster与RedisFanout职责边界.md` + +## 11. 常见问题 + +### 11.1 日志仍然是 standalone + +原因通常是 `cluster_self` 为空或配置文件里的 `nodes` 为空。 + +处理： + +1. 确认 Pod 环境变量 `CLUSTER_SELF`。 +2. 确认启动参数包含 `--cluster_self=$CLUSTER_SELF`。 +3. 确认容器读取的是带完整 `cluster_config.nodes` 的配置。 + +### 11.2 Pod 能启动，但跨 Pod 消息还是不通 + +原因通常是 cluster 内部端口不通，或节点地址写错。 + +处理： + +1. 从任意 Pod `nc -vz 12000`。 +2. 检查 Headless Service selector 是否匹配 Pod label。 +3. 检查 NetworkPolicy 是否放通 egress/ingress。 +4. 检查 `cluster_config.nodes.addr` 是否写成了只在 docker compose 里可用的短域名。 + +### 11.3 扩缩容后部分 Topic 异常 + +原生 Cluster 的 Topic master 基于节点集合计算。扩缩容会改变节点集合，可能触发 Topic 重新路由。 + +建议： + +- 生产先固定 3 节点。 +- 不要频繁自动扩缩容 IM Server。 +- 扩缩容必须配合完整压测和重连验证。 +- 如果要从 3 节点扩到 5 节点，先更新所有 Pod 的 `cluster_config.nodes`，再按发布计划滚动。 + +### 11.4 2 节点能不能用 + +2 节点可以启用基础 Cluster，但不建议开启 failover，也不建议生产使用。 + +生产推荐 3 节点起，便于 leader 心跳和故障判断。 + +## 12. 运维上线检查清单 + +- [ ] 使用 StatefulSet，不使用普通 Deployment 承载 Cluster 节点身份。 +- [ ] 创建 Headless Service。 +- [ ] 每个 Pod 的 `CLUSTER_SELF` 等于 Pod 名。 +- [ ] `cluster_config.nodes` 包含所有 Pod。 +- [ ] 所有 Pod 使用同一份 `cluster_config.nodes`。 +- [ ] 每个 `nodes.addr` 可从其他 Pod 访问。 +- [ ] 放通 cluster 内部 TCP 端口 `12000`。 +- [ ] 普通 HTTP/WebSocket 入口仍通过 Service/Ingress 负载到所有 Pod。 +- [ ] 只有一个入口负责数据库初始化，或数据库初始化已改为独立 Job。 +- [ ] 启动日志不再出现 `Cluster: running as a standalone server.`。 +- [ ] 跨 Pod P2P、群聊、read/recv、通话状态已验收。 +- [ ] 原生 Cluster 启用后，Topic 相关 Redis Fanout 已按职责边界关闭或降级。 diff --git a/server/kafka_cache_invalidate.go b/server/kafka_cache_invalidate.go index 324883e..d6abb76 100644 --- a/server/kafka_cache_invalidate.go +++ b/server/kafka_cache_invalidate.go @@ -38,8 +38,10 @@ type cacheInvalidatePayload struct { Ranges []types.Range `json:"ranges"` } +var kafkaCacheInvalidateConsumerGroup string + func startKafkaCacheInvalidateConsumer() { - startKafkaConsumer("cache_invalidate", handleKafkaCacheInvalidateEvent) + kafkaCacheInvalidateConsumerGroup = startKafkaBroadcastConsumer("cache_invalidate", handleKafkaCacheInvalidateEvent) } func handleKafkaCacheInvalidateEvent(ctx context.Context, envelope kafkaEnvelope, _ kafka.Message) error { @@ -50,17 +52,18 @@ func handleKafkaCacheInvalidateEvent(ctx context.Context, envelope kafkaEnvelope if tenantID.IsZero() { tenantID = types.TenantID(1) } - consume, claimed, err := beginKafkaConsumeForTopic(tenantID, envelope, "cache_invalidate") + consumerGroup := kafkaCacheInvalidateConsumerGroup + consume, claimed, err := beginKafkaConsumeForTopicWithGroup(tenantID, envelope, "cache_invalidate", consumerGroup) if err != nil { return err } if !claimed { - logs.Info.Printf("kafka cache invalidate skipped: event=%s group=%s status=%s", envelope.EventID, kafkaBus.config.ConsumerGroup, consume.Status) + logs.Info.Printf("kafka cache invalidate skipped: event=%s group=%s status=%s", envelope.EventID, consumerGroup, consume.Status) return nil } if err = executeKafkaCacheInvalidateEvent(envelope); err != nil { reason := kafkaCommandErrorReason(err) - failed, markErr := store.HTTPInfrastructure.MarkKafkaConsumeFailed(tenantID, envelope.EventID, kafkaBus.config.ConsumerGroup, reason, err.Error(), time.Now().UTC().Round(time.Millisecond)) + failed, markErr := store.HTTPInfrastructure.MarkKafkaConsumeFailed(tenantID, envelope.EventID, consumerGroup, reason, err.Error(), time.Now().UTC().Round(time.Millisecond)) if markErr != nil { return markErr } @@ -70,7 +73,7 @@ func handleKafkaCacheInvalidateEvent(ctx context.Context, envelope kafkaEnvelope sleepKafkaRetryBackoff(failed.RetryCount) return err } - return store.HTTPInfrastructure.MarkKafkaConsumeSucceeded(tenantID, envelope.EventID, kafkaBus.config.ConsumerGroup, time.Now().UTC().Round(time.Millisecond)) + return store.HTTPInfrastructure.MarkKafkaConsumeSucceeded(tenantID, envelope.EventID, consumerGroup, time.Now().UTC().Round(time.Millisecond)) } func executeKafkaCacheInvalidateEvent(envelope kafkaEnvelope) error { diff --git a/server/kafka_config_events.go b/server/kafka_config_events.go index 22715ac..ac9b77c 100644 --- a/server/kafka_config_events.go +++ b/server/kafka_config_events.go @@ -34,8 +34,10 @@ type configEventPayload struct { Keys []string `json:"keys"` } +var kafkaConfigEventConsumerGroup string + func startKafkaConfigEventConsumer() { - startKafkaConsumer("config_event", handleKafkaConfigEvent) + kafkaConfigEventConsumerGroup = startKafkaBroadcastConsumer("config_event", handleKafkaConfigEvent) } func handleKafkaConfigEvent(ctx context.Context, envelope kafkaEnvelope, _ kafka.Message) error { @@ -46,17 +48,18 @@ func handleKafkaConfigEvent(ctx context.Context, envelope kafkaEnvelope, _ kafka if tenantID.IsZero() { tenantID = types.TenantID(1) } - consume, claimed, err := beginKafkaConsumeForTopic(tenantID, envelope, "config_event") + consumerGroup := kafkaConfigEventConsumerGroup + consume, claimed, err := beginKafkaConsumeForTopicWithGroup(tenantID, envelope, "config_event", consumerGroup) if err != nil { return err } if !claimed { - logs.Info.Printf("kafka config event skipped: event=%s group=%s status=%s", envelope.EventID, kafkaBus.config.ConsumerGroup, consume.Status) + logs.Info.Printf("kafka config event skipped: event=%s group=%s status=%s", envelope.EventID, consumerGroup, consume.Status) return nil } if err = executeKafkaConfigEvent(envelope); err != nil { reason := kafkaCommandErrorReason(err) - failed, markErr := store.HTTPInfrastructure.MarkKafkaConsumeFailed(tenantID, envelope.EventID, kafkaBus.config.ConsumerGroup, reason, err.Error(), time.Now().UTC().Round(time.Millisecond)) + failed, markErr := store.HTTPInfrastructure.MarkKafkaConsumeFailed(tenantID, envelope.EventID, consumerGroup, reason, err.Error(), time.Now().UTC().Round(time.Millisecond)) if markErr != nil { return markErr } @@ -66,7 +69,7 @@ func handleKafkaConfigEvent(ctx context.Context, envelope kafkaEnvelope, _ kafka sleepKafkaRetryBackoff(failed.RetryCount) return err } - return store.HTTPInfrastructure.MarkKafkaConsumeSucceeded(tenantID, envelope.EventID, kafkaBus.config.ConsumerGroup, time.Now().UTC().Round(time.Millisecond)) + return store.HTTPInfrastructure.MarkKafkaConsumeSucceeded(tenantID, envelope.EventID, consumerGroup, time.Now().UTC().Round(time.Millisecond)) } func executeKafkaConfigEvent(envelope kafkaEnvelope) error { @@ -141,10 +144,20 @@ func beginKafkaConsumeForTopic(tenantID types.TenantID, envelope kafkaEnvelope, if kafkaBus == nil { return nil, false, types.ErrUnsupported } + return beginKafkaConsumeForTopicWithGroup(tenantID, envelope, topicName, kafkaBus.config.ConsumerGroup) +} + +func beginKafkaConsumeForTopicWithGroup(tenantID types.TenantID, envelope kafkaEnvelope, topicName, consumerGroup string) (*types.KafkaConsumeRecord, bool, error) { + if kafkaBus == nil { + return nil, false, types.ErrUnsupported + } + if consumerGroup = strings.TrimSpace(consumerGroup); consumerGroup == "" { + consumerGroup = kafkaBus.config.ConsumerGroup + } deadline := time.Now().UTC().Add(5 * time.Minute).Round(time.Millisecond) return store.HTTPInfrastructure.BeginKafkaConsume(tenantID, &types.KafkaConsumeRecord{ EventID: envelope.EventID, - ConsumerGroup: kafkaBus.config.ConsumerGroup, + ConsumerGroup: consumerGroup, SourceTopic: kafkaTopic(topicName), EventType: envelope.EventType, BizType: envelope.BizType, diff --git a/server/kafka_runtime.go b/server/kafka_runtime.go index 69be8e4..b5baeed 100644 --- a/server/kafka_runtime.go +++ b/server/kafka_runtime.go @@ -5,6 +5,7 @@ import ( "crypto/tls" "encoding/json" "errors" + "regexp" "strings" "sync" "time" @@ -88,6 +89,8 @@ type kafkaRuntime struct { var kafkaBus *kafkaRuntime +var kafkaConsumerGroupUnsafeChars = regexp.MustCompile(`[^A-Za-z0-9._-]+`) + func initKafkaRuntime(raw json.RawMessage) error { if len(raw) == 0 { kafkaBus = nil @@ -272,6 +275,19 @@ func publishKafkaEnvelope(ctx context.Context, topicName string, key string, env type kafkaMessageHandler func(context.Context, kafkaEnvelope, kafka.Message) error func startKafkaConsumer(topicName string, handler kafkaMessageHandler) { + startKafkaConsumerWithGroup(topicName, "", handler) +} + +func startKafkaBroadcastConsumer(topicName string, handler kafkaMessageHandler) string { + if kafkaBus == nil || handler == nil { + return "" + } + groupID := kafkaBroadcastConsumerGroup(topicName) + startKafkaConsumerWithGroup(topicName, groupID, handler) + return groupID +} + +func startKafkaConsumerWithGroup(topicName, groupID string, handler kafkaMessageHandler) { if kafkaBus == nil || handler == nil { return } @@ -279,8 +295,11 @@ func startKafkaConsumer(topicName string, handler kafkaMessageHandler) { if topic == "" { topic = topicName } - groupID := kafkaBus.config.ConsumerGroup - if topicName != "" { + groupOverride := groupID != "" + if groupID == "" { + groupID = kafkaBus.config.ConsumerGroup + } + if topicName != "" && !groupOverride { groupID += "." + topicName } reader := kafka.NewReader(kafka.ReaderConfig{ @@ -324,6 +343,22 @@ func startKafkaConsumer(topicName string, handler kafkaMessageHandler) { logs.Info.Printf("Kafka consumer started: topic=%s group=%s", topic, groupID) } +func kafkaBroadcastConsumerGroup(topicName string) string { + if kafkaBus == nil { + return "" + } + base := strings.TrimSpace(kafkaBus.config.ConsumerGroup) + if base == "" { + base = "im-server" + } + nodeID := strings.TrimSpace(currentNodeID()) + if nodeID == "" { + nodeID = "local" + } + nodeID = kafkaConsumerGroupUnsafeChars.ReplaceAllString(nodeID, "-") + return base + "." + topicName + "." + nodeID +} + func (rt *kafkaRuntime) dialer() *kafka.Dialer { dialer := &kafka.Dialer{ ClientID: rt.config.ClientID,