From 2b631ab1763d44d5c055a7eebe108ab5892d105b Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Wed, 30 Sep 2026 17:10:13 +0800 Subject: [PATCH] fix: harden websocket heartbeat under load --- deploy/argocd/im-server-test/README.md | 2 + .../statefulset-performance-patch.yaml | 4 ++ docker/tinode/config.template | 2 + docker/tinode/entrypoint.sh | 2 + docs/test/IM-单机60%容量压力测试方案.md | 4 +- .../IM-单机60%容量压力测试结果-20260929.md | 51 +++++++++++++++++ loadtest/im-pressure/wsclient.go | 25 ++++++++- loadtest/im-pressure/wsclient_test.go | 55 +++++++++++++++++++ server/hdl_websock.go | 34 ++++++++++-- server/hdl_websock_config_test.go | 29 ++++++++++ server/main.go | 8 +++ server/metrics/metrics.go | 21 ++++++- 12 files changed, 228 insertions(+), 9 deletions(-) create mode 100644 server/hdl_websock_config_test.go diff --git a/deploy/argocd/im-server-test/README.md b/deploy/argocd/im-server-test/README.md index a161785..dfe91ca 100644 --- a/deploy/argocd/im-server-test/README.md +++ b/deploy/argocd/im-server-test/README.md @@ -34,5 +34,7 @@ Expected values are requests `1 CPU / 512Mi`, limits `2 CPU / 1Gi`, and liveness path `/livez`. The MySQL pool environment must be `MYSQL_MAX_OPEN_CONNS=24`, `MYSQL_MAX_IDLE_CONNS=8`, `MYSQL_REGISTRY_MAX_OPEN_CONNS=4`, and `MYSQL_REGISTRY_MAX_IDLE_CONNS=2`. +The WebSocket heartbeat environment must be `WS_PING_PERIOD_SECONDS=15` and +`WS_PONG_WAIT_SECONDS=90`. The application repository's deploy workflow waits for reconciliation and fails if ArgoCD restores different values. diff --git a/deploy/argocd/im-server-test/statefulset-performance-patch.yaml b/deploy/argocd/im-server-test/statefulset-performance-patch.yaml index 6237700..a68060c 100644 --- a/deploy/argocd/im-server-test/statefulset-performance-patch.yaml +++ b/deploy/argocd/im-server-test/statefulset-performance-patch.yaml @@ -9,6 +9,10 @@ spec: containers: - name: im-server env: + - name: WS_PING_PERIOD_SECONDS + value: "15" + - name: WS_PONG_WAIT_SECONDS + value: "90" - name: CLUSTER_PROXY_TO_MASTER_BUFFER value: "4096" - name: CLUSTER_PROXY_TO_MASTER_WORKERS diff --git a/docker/tinode/config.template b/docker/tinode/config.template index 17ccdc6..5443833 100644 --- a/docker/tinode/config.template +++ b/docker/tinode/config.template @@ -5,6 +5,8 @@ "static_mount": "/", "grpc_listen": ":16060", "grpc_keepalive_enabled": true, + "ws_ping_period_seconds": $WS_PING_PERIOD_SECONDS, + "ws_pong_wait_seconds": $WS_PONG_WAIT_SECONDS, "api_key_salt": "$API_KEY_SALT", "max_message_size": 4194304, "max_subscriber_count": 128, diff --git a/docker/tinode/entrypoint.sh b/docker/tinode/entrypoint.sh index 426d3aa..c57d7d3 100644 --- a/docker/tinode/entrypoint.sh +++ b/docker/tinode/entrypoint.sh @@ -13,6 +13,8 @@ else CONFIG=working.config : "${REDIS_KEY_PREFIX:=im}" + : "${WS_PING_PERIOD_SECONDS:=15}" + : "${WS_PONG_WAIT_SECONDS:=90}" : "${CLUSTER_PROXY_TO_MASTER_BUFFER:=4096}" : "${CLUSTER_PROXY_TO_MASTER_WORKERS:=16}" : "${CLUSTER_PROXY_TO_MASTER_ENQUEUE_TIMEOUT_MS:=300}" diff --git a/docs/test/IM-单机60%容量压力测试方案.md b/docs/test/IM-单机60%容量压力测试方案.md index 2e42e21..ec40eff 100644 --- a/docs/test/IM-单机60%容量压力测试方案.md +++ b/docs/test/IM-单机60%容量压力测试方案.md @@ -146,6 +146,8 @@ mkdir -p "loadtest/im-pressure/results/${RUN_ID}" 同时采集 Redis、Kafka、MySQL、Pod CPU/内存/网络、Kafka consumer lag、MySQL 连接池等指标。如本机没有 `node-exporter :9100`，删除 `node` URL，但不影响 agent 指标采集。 +发版后必须在服务端 `/metrics` 确认 `im_ws_heartbeat_seconds{setting="ping_period"}=15`、`im_ws_heartbeat_seconds{setting="pong_wait"}=90`，并采集 `im_ws_heartbeat_total{event="pong_received"}` 和 `im_ws_disconnect_total{reason="timeout"}` 的分 Pod 增量。 + ## 7. 报告和独立落库核验 ```bash @@ -176,7 +178,7 @@ export TENANT='006' # 必须与 config.single-machine.json 中的 tenant 一致 ## 9. 单机特有风险 1. 两个 agent 共用一台机器的 CPU、内存、FD、NAT 和出口带宽，测到的是“单机压测端 + 服务端”综合上限，不能代替双机最终验收。 -2. 正式前做 120/600/1200 连接三档阶梯，观察 agent CPU、`go_goroutines`、FD、接收速率和 Ping/Pong 断连，确认压测端本身没有先饱和。当前工具已将投递观测按 `client_mid` 分片、接收账号顺序状态独立加锁，并使用原子延迟直方图；仍需用指标证明本机有余量。 +2. 正式前做 120/600/1200 连接三档阶梯，观察 agent CPU、`go_goroutines`、FD、接收速率和 Ping/Pong 断连，确认压测端本身没有先饱和。当前工具已将投递观测按 `client_mid` 分片、接收账号顺序状态独立加锁，使用原子延迟直方图，并将数据统计回调从 WebSocket socket 读循环解耦；仍需用指标证明本机有余量。 3. 两个 agent 必须使用不同指标端口，不能同时绑定 `:9105`。 4. 压测域名如被 VPN/TUN 接管，起压前必须确认所有当前 IP 的路由都指向物理网卡。 5. 单机 60% 通过后只能进入双机 100% 验收，不能直接得出 2000 连接容量通过的结论。 diff --git a/docs/test/IM-单机60%容量压力测试结果-20260929.md b/docs/test/IM-单机60%容量压力测试结果-20260929.md index bb4c6b3..c972cb2 100644 --- a/docs/test/IM-单机60%容量压力测试结果-20260929.md +++ b/docs/test/IM-单机60%容量压力测试结果-20260929.md @@ -296,3 +296,54 @@ ACK 延迟按 3 个拨号 IP 分组后同步上升，不集中于单一入口。 - `loadtest/im-pressure/results/s600m-pushfix-r2-0930/` - `loadtest/im-pressure/results/s600m-pushfix-r3-0930/` - `loadtest/im-pressure/results/s600m-pushfix-r4-0930/` + +## 16. 异步推送发版后精确 900 人复测 + +Run ID：`s900m-pushfix-r2-0930`，精确群成员 900 人，A/B 各 450 条 WebSocket 连接。两端均完整运行至 5 分钟沉淀、历史回查和 240 秒渐进关闭结束，没有提前终止。 + +| 指标 | 结果 | +|---|---:| +| 就绪连接 | 900 / 900 | +| 发送尝试 / 收到 ACK | 900 / 590 | +| 发送结果未知 | 310（186 条已被对端 WS 观察，124 条未入服） | +| 服务端实际接收并落库 | 776 / 900 | +| ACK P95 | 17075.685ms | +| ACK 台账实时投递 | 381424 / 531000 | +| ACK 台账实时到达率 | 71.8313% | +| 实时缺口 | 149576，全部可从历史补回 | +| 重复 / 乱序 | 0 / 0 | +| 历史回查 | 776 / 776 | +| MySQL 精确命中 | 776 / 776 | +| MySQL 缺失 / 错配 / 重复 / 额外 | 0 / 0 / 0 / 0 | +| 断连 | 410（322 reset、88 abnormal close） | +| Pod 重启 | 0 | + +本轮消息正确性结论很明确：590 条有 ACK，186 条虽然 ACK 超时或 socket 失败，但已被其他 WS 客户端观察到；这 776 条全部能从历史查到，且 MySQL 精确落库 776 条。服务端已接收消息的持久化丢失率为 0，无重复、无字段错配。余下 124 条在连接断开期间未进入服务端，不属于服务端落库后丢失。 + +服务端在 776 条实际入服消息上没有出现容量积压： + +- Topic queue wait P95 不超过 10ms，95.7% 不超过 10ms。 +- push dispatch 776/776 成功入队，无 full；队列等待 P95 不超过 5ms，dispatch P95 不超过 25ms。 +- WS send queue 等待 P95 不超过 5ms，socket text write P95 不超过 0.5ms；1177345 次成功写入，16 次写入失败均是对端已关闭连接。 +- 3 个 Pod 的 Kafka consumer lag 和 oldest backlog age 始终为 0，无 Topic、push dispatch 或 WS send queue full。 +- 渐进关闭后 push dispatch 和 WS send queue 深度均为 0，`topic_presence_pending` 回落为 0。 + +压测期间客户端记录 410 次非预期断连，服务端指标同期新增 412 次 WS timeout，三个 Pod 分别为 135、133、144 次，并非单一 Pod 故障。客户端 reset 样本同样均匀分布在三个公网 IP。压测 agent A/B 峰值 CPU 约为 20.8%/22.0% 单核，峰值 RSS 约为 108/101MiB，FD 均为 460，本机进程没有出现 CPU、内存或 FD 容量耗尽。结合本章前已观测到的高 RTT、ICMP 丢包和 HTTPS TTFB，本轮只能判定“当前端到端公网路径下 900 长连接门禁失败”，具体故障点可能在压测机网络、NAT/ISP、Ingress/LB 或其间 TCP 链路，不能据此判定 IM 服务端在 900 人处理容量上失败。 + +下一次 900 人容量验收必须改为集群同地域压测机，或先取得 Ingress/LB 的 reset、TCP 重传、出站带宽和 upstream 时延指标。网络基线不合格时，继续从当前机器反复施压只会重复得到链路上限，无法给出服务端容量上限。 + +新增产物： + +- `loadtest/im-pressure/results/s900m-pushfix-r2-0930/report.json` +- `loadtest/im-pressure/results/s900m-pushfix-r2-0930/db-verification.json` +- `loadtest/im-pressure/results/s900m-pushfix-r2-0930/server-metrics/` + +## 17. 900 连接心跳容错与压测端解耦（待发版复测） + +针对 R2 在约一个心跳超时窗口后出现的 410 次断连，已完成三项改造： + +1. WebSocket Ping 周期和 Pong 等待时间从硬编码中拆出，支持 `ws_ping_period_seconds` 和 `ws_pong_wait_seconds` 配置。默认为 15 秒 Ping、90 秒 Pong 容错，不再复用 45 秒业务 idle timeout。 +2. 新增 `im_ws_heartbeat_seconds` 和 `im_ws_heartbeat_total{event="pong_received"}`，用于核对每个 Pod 的有效心跳配置和 Pong 接收数；测试环境 ArgoCD patch 显式设置 15/90 秒。 +3. 压测客户端为每条连接新增有序数据回调队列。Socket 读循环不再同步执行消息去重、顺序和延迟统计，避免统计锁竞争阻塞 ACK、Ping 和 Pong 读取。 + +发版后仍使用同一精确 900 人群复测。起压前先确认三个 Pod 的 heartbeat gauge 均为 15/90；验收时对比 Pong 增量、timeout 增量、agent 断连和入服/落库数。若仍然同时批量 timeout，下一步必须取得 Ingress/LB TCP 指标或改用同地域压测机，不再继续放大应用心跳窗口。 diff --git a/loadtest/im-pressure/wsclient.go b/loadtest/im-pressure/wsclient.go index 9bbc454..0949b34 100644 --- a/loadtest/im-pressure/wsclient.go +++ b/loadtest/im-pressure/wsclient.go @@ -40,6 +40,8 @@ type ctrlError struct { Text string } +const dataCallbackQueueSize = 512 + func (e *ctrlError) Error() string { return fmt.Sprintf("ctrl code=%d text=%s", e.Code, e.Text) } @@ -56,6 +58,7 @@ type WSClient struct { seq uint64 waits map[string]chan ctrlPacket done chan struct{} + data chan dataPacket closed atomic.Bool } @@ -84,6 +87,7 @@ func connectWSClient(ctx context.Context, cfg *Config, account AccountRecord, ap account: account, conn: conn, commandTimeout: timeout, onData: onData, onDisconnect: onDisconnect, waits: make(map[string]chan ctrlPacket), done: make(chan struct{}), } + client.startDataProcessor() go client.readLoop() if _, err = client.command(ctx, "hi", map[string]any{"hi": map[string]any{ "id": client.nextID("hi"), "ver": "0.25", "ua": "im-pressure/1.0", @@ -346,6 +350,9 @@ func (c *WSClient) removeWait(id string) { func (c *WSClient) readLoop() { defer func() { _ = c.conn.Close() + if c.data != nil { + close(c.data) + } close(c.done) }() for { @@ -372,12 +379,28 @@ func (c *WSClient) readLoop() { if raw := envelope["data"]; len(raw) > 0 && c.onData != nil { var data dataPacket if json.Unmarshal(raw, &data) == nil { - c.onData(c.account, data) + if c.data == nil { + c.onData(c.account, data) + } else { + c.data <- data + } } } } } +func (c *WSClient) startDataProcessor() { + if c.onData == nil || c.data != nil { + return + } + c.data = make(chan dataPacket, dataCallbackQueueSize) + go func() { + for data := range c.data { + c.onData(c.account, data) + } + }() +} + func (c *WSClient) startKeepalive(interval time.Duration) { go func() { ticker := time.NewTicker(interval) diff --git a/loadtest/im-pressure/wsclient_test.go b/loadtest/im-pressure/wsclient_test.go index 6e98984..5082ac1 100644 --- a/loadtest/im-pressure/wsclient_test.go +++ b/loadtest/im-pressure/wsclient_test.go @@ -84,3 +84,58 @@ func TestWSClientReadLoopClosesConnectionAfterDecodeError(t *testing.T) { t.Fatal("read loop exited without closing the underlying connection") } } + +func TestWSClientDataCallbackDoesNotBlockPong(t *testing.T) { + pong := make(chan struct{}, 1) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := (&websocket.Upgrader{}).Upgrade(w, r, nil) + if err != nil { + return + } + defer conn.Close() + conn.SetPongHandler(func(string) error { + select { + case pong <- struct{}{}: + default: + } + return nil + }) + if err = conn.WriteJSON(map[string]any{"data": map[string]any{ + "topic": "grp-test", "seq": 1, "head": map[string]any{}, "content": map[string]any{}, + }}); err != nil { + return + } + if err = conn.WriteControl(websocket.PingMessage, nil, time.Now().Add(time.Second)); err != nil { + return + } + conn.SetReadDeadline(time.Now().Add(time.Second)) + for { + if _, _, readErr := conn.ReadMessage(); readErr != nil { + return + } + } + })) + defer server.Close() + + conn, _, err := websocket.DefaultDialer.Dial("ws"+strings.TrimPrefix(server.URL, "http"), nil) + if err != nil { + t.Fatal(err) + } + blocked := make(chan struct{}) + client := &WSClient{ + conn: conn, waits: make(map[string]chan ctrlPacket), done: make(chan struct{}), + onData: func(AccountRecord, dataPacket) { <-blocked }, + } + client.startDataProcessor() + go client.readLoop() + defer func() { + close(blocked) + client.Close() + }() + + select { + case <-pong: + case <-time.After(2 * time.Second): + t.Fatal("pong was blocked by the data callback") + } +} diff --git a/server/hdl_websock.go b/server/hdl_websock.go index 61ce16d..194288c 100644 --- a/server/hdl_websock.go +++ b/server/hdl_websock.go @@ -28,14 +28,35 @@ import ( const ( // Time allowed to write a message to the peer. writeWait = 10 * time.Second - - // Time allowed to read the next pong message from the peer. - pongWait = idleSessionTimeout - - // Send pings to peer with this period. Must be less than pongWait. - pingPeriod = 15 * time.Second + // Heartbeat defaults tolerate transient mobile and public-network stalls while + // still detecting dead connections in a bounded amount of time. + defaultWSPongWait = 90 * time.Second + defaultWSPingPeriod = 15 * time.Second ) +var ( + pongWait = defaultWSPongWait + pingPeriod = defaultWSPingPeriod +) + +func configureWSHeartbeat(pingSeconds, pongSeconds int) error { + ping := defaultWSPingPeriod + pong := defaultWSPongWait + if pingSeconds > 0 { + ping = time.Duration(pingSeconds) * time.Second + } + if pongSeconds > 0 { + pong = time.Duration(pongSeconds) * time.Second + } + if ping >= pong { + return errors.New("websocket ping period must be shorter than pong wait") + } + pingPeriod = ping + pongWait = pong + immetrics.SetWSHeartbeatConfig(ping, pong) + return nil +} + func (sess *Session) closeWS() { if sess.proto == WEBSOCK { sess.ws.Close() @@ -52,6 +73,7 @@ func (sess *Session) readLoop() { sess.ws.SetReadDeadline(time.Now().Add(pongWait)) sess.ws.SetPongHandler(func(string) error { atomic.StoreInt64(&sess.lastAction, types.TimeNow().UnixNano()) + immetrics.IncWSHeartbeat("pong_received") sess.refreshPresence("ws.pong") sess.ws.SetReadDeadline(time.Now().Add(pongWait)) return nil diff --git a/server/hdl_websock_config_test.go b/server/hdl_websock_config_test.go new file mode 100644 index 0000000..70f17bd --- /dev/null +++ b/server/hdl_websock_config_test.go @@ -0,0 +1,29 @@ +package main + +import ( + "testing" + "time" +) + +func TestConfigureWSHeartbeat(t *testing.T) { + originalPing, originalPong := pingPeriod, pongWait + t.Cleanup(func() { + pingPeriod, pongWait = originalPing, originalPong + }) + + if err := configureWSHeartbeat(20, 120); err != nil { + t.Fatal(err) + } + if pingPeriod != 20*time.Second || pongWait != 120*time.Second { + t.Fatalf("heartbeat ping=%s pong=%s", pingPeriod, pongWait) + } + if err := configureWSHeartbeat(90, 90); err == nil { + t.Fatal("expected invalid heartbeat configuration to fail") + } + if err := configureWSHeartbeat(0, 0); err != nil { + t.Fatal(err) + } + if pingPeriod != defaultWSPingPeriod || pongWait != defaultWSPongWait { + t.Fatalf("default heartbeat ping=%s pong=%s", pingPeriod, pongWait) + } +} diff --git a/server/main.go b/server/main.go index d6f893c..99998b0 100644 --- a/server/main.go +++ b/server/main.go @@ -322,6 +322,10 @@ type configType struct { // If true, do not attempt to negotiate websocket per message compression (RFC 7692.4). // It should be disabled (set to true) if you are using MSFT IIS as a reverse proxy. WSCompressionDisabled bool `json:"ws_compression_disabled"` + // WebSocket heartbeat period and the maximum time to wait for a Pong, in seconds. + // Zero values use the built-in defaults. + WSPingPeriodSeconds int `json:"ws_ping_period_seconds"` + WSPongWaitSeconds int `json:"ws_pong_wait_seconds"` // Address:port to listen for gRPC clients. If blank gRPC support will not be initialized. // Could be overridden from the command line with --grpc_listen. GrpcListen string `json:"grpc_listen"` @@ -716,6 +720,10 @@ func main() { // Websocket compression. globals.wsCompression = !config.WSCompressionDisabled + if err = configureWSHeartbeat(config.WSPingPeriodSeconds, config.WSPongWaitSeconds); err != nil { + logs.Err.Fatal("Invalid WebSocket heartbeat configuration: ", err) + } + logs.Info.Printf("WebSocket heartbeat ping=%s pong_wait=%s", pingPeriod, pongWait) if config.Media != nil { if config.Media.UseHandler == "" { diff --git a/server/metrics/metrics.go b/server/metrics/metrics.go index 289b08d..bdebd10 100644 --- a/server/metrics/metrics.go +++ b/server/metrics/metrics.go @@ -71,6 +71,14 @@ var ( Name: "im_ws_disconnect_total", Help: "WebSocket read-loop disconnects by pod and bounded reason.", }, []string{"pod", "reason"}) + IMWSHeartbeatTotal = prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "im_ws_heartbeat_total", + Help: "WebSocket heartbeat events by pod and event.", + }, []string{"pod", "event"}) + IMWSHeartbeatSeconds = prometheus.NewGaugeVec(prometheus.GaugeOpts{ + Name: "im_ws_heartbeat_seconds", + Help: "Effective WebSocket heartbeat timing by pod and setting.", + }, []string{"pod", "setting"}) IMWSSendQueueCurrent = prometheus.NewGaugeVec(prometheus.GaugeOpts{ Name: "im_ws_send_queue_current", Help: "Current number of queued WebSocket outbound channel entries across sessions on a pod.", @@ -216,7 +224,8 @@ var ( func init() { Registry.MustRegister(WSConnectionsActive, IMWSConnectionsActive, IMWSWriteTotal, IMWSWriteDuration, - IMWSWriteBytesTotal, IMWSDisconnectTotal, IMWSSendQueueCurrent, IMWSSendQueueCapacity, + IMWSWriteBytesTotal, IMWSDisconnectTotal, IMWSHeartbeatTotal, IMWSHeartbeatSeconds, + IMWSSendQueueCurrent, IMWSSendQueueCapacity, IMWSSendQueueEnqueueTotal, IMWSSendQueueDepth, IMWSSendQueueWaitDuration, IMClusterLivePeers, IMClusterExpectedPeers, HTTPRequestsTotal, HTTPRequestDuration, DBQueryDuration, @@ -332,6 +341,16 @@ func IncWSDisconnect(reason string) { IMWSDisconnectTotal.WithLabelValues(metricPodName(), normalizeLabel(reason)).Inc() } +func IncWSHeartbeat(event string) { + IMWSHeartbeatTotal.WithLabelValues(metricPodName(), normalizeLabel(event)).Inc() +} + +func SetWSHeartbeatConfig(pingPeriod, pongWait time.Duration) { + pod := metricPodName() + IMWSHeartbeatSeconds.WithLabelValues(pod, "ping_period").Set(pingPeriod.Seconds()) + IMWSHeartbeatSeconds.WithLabelValues(pod, "pong_wait").Set(pongWait.Seconds()) +} + func ObserveWSSendQueueEnqueue(depth, capacity int, accepted bool) { result := "ok" if !accepted {