diff --git a/docs/test/IM-单聊压测Pod日志根因分析与改进方案-20260927.md b/docs/test/IM-单聊压测Pod日志根因分析与改进方案-20260927.md index 17e3dcf..e509be0 100644 --- a/docs/test/IM-单聊压测Pod日志根因分析与改进方案-20260927.md +++ b/docs/test/IM-单聊压测Pod日志根因分析与改进方案-20260927.md @@ -349,3 +349,28 @@ Run ID:`p2p-200-20260927-04`。ArgoCD Application `im-server-test` 的真实 Kafka `im-server.message_persist` lag 在发送后峰值为 2946,并在发送结束后约 113 秒归零,超过 60 秒目标。历史 API 和 MySQL 最终均确认 4000 条完整落库,说明这是落库延迟而不是消息丢失。进入 500 对测试前,应优先检查 message persist consumer 的批量写入、worker/partition 并行度和 MySQL 写入耗时。 连接集中关闭期间,三个 Pod 分别记录 10、14、22 次 `failed to close login audit ... Deadlock found`,共 46 次。该问题未影响消息表,但会导致登录审计结束状态不完整。登录审计关闭更新应增加 MySQL 1213/1205 有界重试,并检查更新条件和索引以缩小锁范围。 + +## 14. 持久化并行与审计死锁修复后的第五轮复测 + +Run ID:`p2p-200-20260927-05`。本轮已将 `im.message.persist` 和 `im.group.message.persist` 固化为 3 分区、3 副本;同一 Topic 继续作为 Kafka key,保证会话内顺序。服务端同时按 batch 内最大 Seq 合并 Topic 和发送者已读位点更新,并为登录审计关闭增加复合索引和 MySQL 死锁重试。 + +| 指标 | 结果 | +|---|---:| +| WS Ready | 400/400 | +| Publish 尝试 / ACK | 4000 / 4000 | +| ACK 后应到达 / 实际到达 | 4000 / 4000 | +| 丢失 / 重复 / 乱序 / 断连 | 0 / 0 / 0 / 0 | +| 历史查询 / MySQL 精确匹配 | 4000/4000 / 4000/4000 | +| MySQL 缺失 / 错配 / 重复 / 非 ACK 落库 | 0 / 0 / 0 / 0 | +| 总体精确 ACK P50 / P95 / P99 / Max | 375.314 / 547.516 / 670.317 / 2845.347 ms | +| A 机精确 ACK P95 | 408.155 ms | +| B 机精确 ACK P95 | 594.533 ms | +| Pod 0/1/2 Kafka 峰值 lag | 676 / 251 / 143 | +| 最慢 Kafka 分区归零时间 | 发送结束后约 52.5s | +| Cluster 队列峰值 / CPU throttled 增量 | 0 / 三 Pod 均为 0 | +| 登录审计关闭死锁 | 三 Pod 均为 0 | +| Kafka batch failure / Pod 重启 | 0 / 0 | + +结论:三分区已由三个 Pod 实际并行消费,单 Pod 峰值 lag 从第四轮的 2946 降至 676,最慢消费者在 60 秒门槛内归零。消息正确性、最终落库、会话内顺序、CPU 限流和登录审计死锁均通过。 + +报告仍因 ACK P95 门禁显示失败。直方图只能给出 1000ms 上界,按 4000 条 ACK 原始时间戳和 nearest-rank 口径计算的实际 P95 为 547.516ms。A 机已经低于 500ms,而 B 机为 594.533ms,且 B 机存在 2.845s 最大长尾;两台机器使用相同公共入口和相同服务端版本,因此继续修改服务端前,应先采集 B 机到 `api.118im.com` 的 DNS、TCP、TLS、WebSocket RTT 与时钟偏差,并使用定向 Pod 入口做 A/B 对照,区分公网链路长尾和服务端处理耗时。 diff --git a/loadtest/im-pressure/merge.go b/loadtest/im-pressure/merge.go index 3482325..b40982c 100644 --- a/loadtest/im-pressure/merge.go +++ b/loadtest/im-pressure/merge.go @@ -87,7 +87,7 @@ func runMerge(aPath, bPath, output string) error { if report.PublishAttempted > 0 { report.PublishSuccessRate = float64(report.PublishAccepted) / float64(report.PublishAttempted) } - report.AckP95MS = mergedHistogramP95(report.Agents, func(summary AgentSummary) HistogramSnapshot { return summary.AckLatency }) + report.AckP95MS = mergedAckP95(report.Agents) report.DeliveryP95MS = mergedHistogramP95(report.Agents, func(summary AgentSummary) HistogramSnapshot { return summary.DeliveryLatency }) if report.ExpectedDeliveries > 0 { report.ArrivalRate = float64(report.ReceivedDeliveries) / float64(report.ExpectedDeliveries) @@ -102,7 +102,7 @@ func runMerge(aPath, bPath, output string) error { fmt.Sprintf("publish success rate=%.6f expected>=0.999900", report.PublishSuccessRate)) } if report.AckP95MS > 500 { - report.FailureReasons = append(report.FailureReasons, fmt.Sprintf("ack p95=%.0fms expected<=500ms", report.AckP95MS)) + report.FailureReasons = append(report.FailureReasons, fmt.Sprintf("ack p95=%.3fms expected<=500ms", report.AckP95MS)) } if report.DeliveryP95MS > 1000 { report.FailureReasons = append(report.FailureReasons, @@ -121,6 +121,26 @@ func runMerge(aPath, bPath, output string) error { return writeJSONFile(output, report) } +func mergedAckP95(agents []AgentSummary) float64 { + var expected int64 + latencies := make([]float64, 0) + for _, agent := range agents { + expected += agent.AckLatency.Count + for _, message := range agent.AcceptedLocal { + if message.SentAt.IsZero() || message.AckAt.Before(message.SentAt) { + continue + } + latencies = append(latencies, float64(message.AckAt.Sub(message.SentAt))/float64(time.Millisecond)) + } + } + if expected == 0 || int64(len(latencies)) != expected { + return mergedHistogramP95(agents, func(summary AgentSummary) HistogramSnapshot { return summary.AckLatency }) + } + sort.Float64s(latencies) + index := (len(latencies)*95 + 99) / 100 + return latencies[index-1] +} + func mergedHistogramP95(agents []AgentSummary, selectHistogram func(AgentSummary) HistogramSnapshot) float64 { var count int64 buckets := make([]int64, len(latencyBounds)+1) diff --git a/loadtest/im-pressure/merge_test.go b/loadtest/im-pressure/merge_test.go index 3c2d0c9..eb52137 100644 --- a/loadtest/im-pressure/merge_test.go +++ b/loadtest/im-pressure/merge_test.go @@ -5,6 +5,7 @@ import ( "os" "path/filepath" "testing" + "time" ) func TestRunMergeComputesArrivalAndPersistenceGate(t *testing.T) { @@ -53,6 +54,32 @@ func TestMergedHistogramP95(t *testing.T) { } } +func TestMergedAckP95UsesExactAcceptedTimestamps(t *testing.T) { + base := time.Date(2026, 9, 27, 8, 0, 0, 0, time.UTC) + agents := []AgentSummary{{ + AckLatency: HistogramSnapshot{Count: 4, MaxMS: 900, Buckets: map[string]int64{"1s": 4}}, + AcceptedLocal: []AcceptedMessage{ + {SentAt: base, AckAt: base.Add(100 * time.Millisecond)}, + {SentAt: base, AckAt: base.Add(200 * time.Millisecond)}, + {SentAt: base, AckAt: base.Add(300 * time.Millisecond)}, + {SentAt: base, AckAt: base.Add(400 * time.Millisecond)}, + }, + }} + if got := mergedAckP95(agents); got != 400 { + t.Fatalf("exact p95=%v want=400", got) + } +} + +func TestMergedAckP95FallsBackWhenRawSamplesAreIncomplete(t *testing.T) { + agents := []AgentSummary{{ + AckLatency: HistogramSnapshot{Count: 2, MaxMS: 900, Buckets: map[string]int64{"1s": 2}}, + AcceptedLocal: []AcceptedMessage{{SentAt: time.Now(), AckAt: time.Now().Add(100 * time.Millisecond)}}, + }} + if got := mergedAckP95(agents); got != 1000 { + t.Fatalf("fallback p95=%v want=1000", got) + } +} + func writeSummaryForTest(t *testing.T, path string, summary AgentSummary) { t.Helper() raw, err := json.Marshal(summary)