From d3054f62b530c4e12ac42e2b9761b3a29ac5daff Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Mon, 5 Oct 2026 17:36:18 +0800 Subject: [PATCH] test(load): add idle and tunable group scenarios --- ...M-group2k-r12-双机群聊复测分析-20261005.md | 168 ++++++++++++++++++ loadtest/im-pressure/README.md | 9 +- loadtest/im-pressure/agent.go | 90 ++++++++-- loadtest/im-pressure/agent_history_test.go | 27 +++ loadtest/im-pressure/config.example.json | 9 + .../config.single-machine.example.json | 9 + loadtest/im-pressure/main.go | 2 +- loadtest/im-pressure/model.go | 114 +++++++++--- 8 files changed, 385 insertions(+), 43 deletions(-) create mode 100644 docs/test/IM-group2k-r12-双机群聊复测分析-20261005.md diff --git a/docs/test/IM-group2k-r12-双机群聊复测分析-20261005.md b/docs/test/IM-group2k-r12-双机群聊复测分析-20261005.md new file mode 100644 index 0000000..9680749 --- /dev/null +++ b/docs/test/IM-group2k-r12-双机群聊复测分析-20261005.md @@ -0,0 +1,168 @@ +# GROUP-2000 R12 双机群聊复测分析 + +## 1. 结论 + +`group2k-r12-20261004-01` 判定为 **失败，2000 人群聊容量仍未通过**。 + +本轮有一个明确的正向结果：R11 后的 Kafka 部分批次修复已经生效。服务端实际接收并可从投递或历史中确认的 1694 条消息，MySQL 精确找到 1694 条，缺失、错配、重复和额外记录均为 0。 + +实时链路仍然失败。A/B 虽然均以 1000/1000 条连接起跑，但负载开始约 33 秒后公网 `/metrics` 开始连续超时，随后出现两波大规模 WebSocket reset。最终实时到达率只有 74.739%，ACK P95 为 74.72 秒，包含迟到 ACK 的控制响应 P95 达 407.90 秒。 + +与 R11 不同，本轮 Pod 没有重启，日志也没有再次出现 CoreDNS 超时。证据表明本轮先发生群 Topic 串行处理积压和入口响应失速，连接被成批 reset 后又触发重连风暴，三个 Pod 的业务库连接池均出现约 9,200-11,350 秒累计等待，最终形成放大回路。 + +## 2. 核心结果 + +| 指标 | R12 结果 | 判定 | +| --- | ---: | --- | +| WS 起跑就绪 | 2000/2000 | 通过 | +| A/B 起跑差 | 约 33ms | 通过 | +| 发送尝试 | 2000 | A/B 各 1000 | +| 最终明确 ACK | 1501 | 75.05%，失败 | +| 发送结果未知 | 499 | 失败 | +| 服务端接收/回查确认消息 | 1694 | 另有 306 条未确认进入服务端 | +| ACK P95 | 74.72s | 失败 | +| 含迟到 ACK 的控制响应 P95 | 407.90s | 失败 | +| 期望实时投递 | 3,388,000 | 1694 x 2000 | +| 实际实时投递 | 2,532,160 | 到达率 74.739% | +| 实时缺失 | 855,840 | 丢失率 25.261% | +| MySQL 精确核验 | 1694/1694 | 通过 | +| 重复 / 乱序 | 0 / 0 | 已收到部分正确 | +| WS 断连 | 1257 | 失败 | +| Pod 重启 | 0 | 无应用容器重启 | + +`publish_final_accepted=1501` 只统计客户端最终观察到的 ACK。数据库和消息台账共同证明实际有 1694 条进入服务端，因此其中 193 条属于客户端未拿到 ACK、但服务端已经处理的消息。 + +## 3. A/B 分机结果 + +| 指标 | 机器 A | 机器 B | +| --- | ---: | ---: | +| 起跑连接 | 1000 | 1000 | +| 发送尝试 | 1000 | 1000 | +| 及时 ACK | 547 | 601 | +| 迟到 ACK | 327 | 26 | +| 结果未知 | 126 | 373 | +| 收到群消息 | 1,389,624 | 1,142,536 | +| 实时缺失 | 304,376 | 551,464 | +| 断连 | 433 | 824 | +| 主要断连原因 | reset 413、timeout 20 | abnormal close 405、reset 419 | + +B 在第一波连接重置中受影响更重；第二波则同时影响 A/B，不能归因于单台压测机。 + +## 4. 故障时间线 + +时间均为 UTC，北京时间加 8 小时： + +- 15:57:31：A/B 同步起跑，2000 个连接全部就绪。 +- 15:57:34-15:57:41：公网 `/metrics` 已从亚秒上升到约 2.5 秒。 +- 15:58:04：公网 `/metrics` 首次 10 秒超时；此后连续 28 次超时。 +- 15:58:24-15:58:59：第一波约 394 条服务端连接集中关闭，主要对应 B；B 汇总为 405 条 abnormal close。 +- 16:00:15 起：`im-server-0` 开始出现用户缓存更新队列满，窗口内共 78 次。 +- 16:01:50-16:02:10：第二波约 810 条服务端连接集中关闭；A/B 汇总的 reset 合计为 832 条。 +- 16:02:06 起：大量重连登录出现 `context deadline exceeded`。 +- 16:02:08：`im-server-1` readiness 请求本身超时；期间还出现一次 liveness 超时，但未达到重启阈值。 +- 16:02:39-16:02:47：三个 Pod 均产生 readiness 503 事件。 +- 16:02:48、16:02:50：`im-server-2`、`im-server-1` 恢复 Ready；`im-server-0` 未形成持续 NotReady。 +- 16:05:04：公网 `/metrics` 恢复 200，首次恢复请求仍耗时约 4.6 秒。 +- 16:11:55：A/B 汇总完成。 + +三个 Pod 的 `restartCount` 均为 0。R11 出现过的 `10.43.0.10:53`、`no such host`、DNS connection refused 在本轮窗口内均未出现，因此不能继续把本轮失败归因于 CoreDNS。 + +## 5. 容量瓶颈 + +群 Topic master 位于 `im-server-2`。最终指标显示： + +| 指标 | 数值 | +| --- | ---: | +| 群发布处理次数 | 1694 | +| Topic publish 处理总耗时 | 116.59s | +| 单条平均处理耗时 | 约 68.8ms | +| Topic 队列等待总耗时 | 2279.92s | +| 单条平均排队 | 约 1.35s | +| 队列等待最大桶 | 30s 内 | +| master 广播总耗时 / 次数 | 21.33s / 4803 | +| master 单次广播平均耗时 | 约 4.4ms | + +按 68.8ms 的串行 Topic 处理耗时估算，单群持续处理能力约为 14.5 条/秒，低于本轮约 20 条/秒的目标发送速率。共享序列化后的广播本身仍然只有毫秒级，不是主要瓶颈；耗时主要位于 Topic publish 的保存、Kafka produce、权限检查和推送准备等关键路径。 + +这也解释了为何在正式断连前 ACK 已经出现 20-90 秒长尾，并且 `/metrics` 先失去响应。 + +## 6. 断连与入口 + +Pod 日志中的两波连接关闭均来自入口上游连接；客户端看到的是三个源站 IP 返回 `connection reset by peer`，不是主动退出压测程序： + +- 第一波关闭连接的服务端会话寿命平均约 122 秒，集中影响 B。 +- 第二波同时影响 A/B；旧连接寿命主要集中在约 330 秒和 660 秒，说明它是同一时刻的批量重置，而非每条连接独立超时。 +- 三个源 IP 最终汇聚到同一个入口上游地址的现象仍然存在，入口层缺少副本、连接上限、reload/restart、upstream reset 和节点 conntrack 证据。 + +应用负载先失速，入口随后批量 reset，因此当前不能只改客户端重连参数来宣告通过。运维仍需补齐 Traefik、入口节点和 Service endpoint 的同窗口证据，判断 reset 是入口进程、节点网络、连接上限还是上游超时策略触发。 + +## 7. 重连放大与数据库 + +压测前后业务库连接池增量如下： + +| Pod | 新增等待次数 | 新增累计等待 | 平均每次等待 | +| --- | ---: | ---: | ---: | +| im-server-0 | 817 | 9956.29s | 约 12.2s | +| im-server-1 | 770 | 9218.27s | 约 12.0s | +| im-server-2 | 768 | 11348.20s | 约 14.8s | + +每个 Pod 的 `meeting_im` 最大连接数为 24。连接集中断开后，压测端自动重连触发登录、用户状态、IP 白名单和认证版本查询，连接池等待迅速升高；登录进一步超时，readiness 中的数据库检查也随之失败，Pod 被入口摘除，形成反馈回路。 + +不能简单地只提高连接池上限。下一轮前应先定位连接被长时间占用的 SQL/事务，并为登录重连增加服务端限流、排队和抖动；否则只是把压力转移到 MySQL。 + +## 8. Kafka 与落库 + +数据库直连核验结果： + +| 项目 | 结果 | +| --- | ---: | +| tenant | allrich / tenant_id=8 | +| 期望消息 | 1694 | +| 精确找到 | 1694 | +| 缺失 | 0 | +| 字段错配 | 0 | +| 重复行 | 0 | +| 额外记录 | 0 | +| 分表 | `im_008_group_messages_202610`、`im_008_messages` | + +这证明 R11 的 Kafka 部分批次丢弃问题已修复。最后一次公网指标采样中某个 reader 的本地 `im_kafka_consumer_lag` 仍显示 111，但稍后的数据库核验已经完整；再次说明 `kafka-go Reader.Stats().Lag` 不能作为 consumer group 已提交 offset 的最终依据。 + +## 9. 改进优先级 + +### P0：群 Topic 关键路径 + +1. 对 Topic publish 的 save、Kafka produce、权限查询、推送准备分别补充 P95/P99 和并发占用，确认 68.8ms 的主要组成。 +2. 将不要求阻塞 ACK 的推送准备、通知类工作移出 Topic 单写者关键路径。 +3. 对必须保持顺序的持久化和序列分配设计有界流水线；保持同群顺序和失败可重试，不能重新引入已移除的业务 outbox。 +4. 优化后先跑 10、12、15、18、20 msg/s 阶梯，连续两轮确认 Topic 队列不持续增长，再重新判定 20 msg/s 容量。 + +### P0：入口批量 reset + +1. 由运维提取 15:57:30-16:05:10 的 Traefik Pod 日志、事件、restart、reload、upstream reset 和打开连接数。 +2. 核对三个公网源 IP 是否最终只有一个 Traefik 实例承载；入口至少配置两个副本、反亲和和 PDB。 +3. 同步采集节点 conntrack、TCP retransmit、listen overflow、文件描述符、网卡丢包和 CPU throttling。 +4. 下一轮同时探测公网域名、每个源 IP、Service 和三个 Pod，避免 `/metrics` 单点失明。 + +### P1：重连保护 + +1. 客户端维持自动重连，但使用指数退避和随机抖动，避免 2000 条连接同步登录。 +2. 服务端对登录重建增加有界并发和快速过载响应，不让请求无限占用数据库连接池。 +3. 分解登录 SQL 指标，定位连接池长等待对应的查询或事务。 +4. readiness 的数据库检查应使用独立的小连接预算或短路缓存，避免业务池耗尽导致全部 Endpoint 同时被摘除；liveness 继续只检查进程存活。 + +## 10. 下一轮顺序 + +1. 先做 2000 连接、0 发消息的 15 分钟稳定性测试，排除入口固定连接寿命和 B 网络路径问题。 +2. 再做单群阶梯：10、12、15、18、20 msg/s，每档 5 分钟，观察 Topic queue 是否线性增长。 +3. 完成入口和数据库现场采集后，重复 R12 原参数两轮。 +4. 通过标准：2000/2000 连接全程稳定、发送成功率不低于 99.99%、实时到达率不低于 99.99%、DB 精确落库、无批量 reset、无持续 Topic/DB 队列增长。 + +## 11. 产物 + +- 结果目录：`loadtest/im-pressure/results/group2k-r12-20261004-01/` +- 汇总：`report.json` +- 数据库核验：`db-verification.json` +- A/B 明细：`agent-a-summary.json`、`agent-b-summary.json` +- 协调器台账：`coordinator.json` +- 指标：`metrics.jsonl` +- 被测代码：`965f94f fix(kafka): preserve partially fetched message batches` diff --git a/loadtest/im-pressure/README.md b/loadtest/im-pressure/README.md index cf01add..30c2703 100644 --- a/loadtest/im-pressure/README.md +++ b/loadtest/im-pressure/README.md @@ -197,8 +197,13 @@ mkdir -p "loadtest/im-pressure/results/${RUN_ID}" 两个 agent 都建立 1000 条连接后，协调器下发统一的 UTC `start_at`。场景可替换为： - `p2p`：1000 对用户两侧同时发送，默认每人 10 条、间隔 1 秒。 -- `group`：`users_per_machine=1000` 时每个 agent 有 500 个发送者，每人 2 条；全局每 100ms 一条，共 2000 条。其他容量按每个 agent 用户数的 50% 选发送者。 -- `mixed`：`users_per_machine=1000` 时使用 300 对私聊用户和每个 agent 250 个群发送者；其他容量分别按 30% 和 25% 同比计算。 +- `group`：默认每个 agent 有 `group_senders_per_machine` 个发送者，每人 `group_rounds` 条；两机合并后按 `group_publish_interval` 控制全局发送间隔。双机默认 500+500 发送者、100ms 一条，共 2000 条；单机 60% 默认 300+300 发送者。 +- `mixed`：默认使用 `mixed_p2p_users_per_machine` 对私聊用户和 `mixed_group_senders_per_machine` 个群发送者。双机默认每机 300 个私聊用户、250 个群发送者；单机 60% 默认每机 180 个私聊用户、150 个群发送者。 +- `idle`：只建立并维持 WS 连接，不发送业务消息，持续 `idle_duration`。该场景用于先确认 2000 长连接在入口、Pod、DNS、心跳和自动重连下是否稳定。 + +群聊容量阶梯建议只改 `group_publish_interval`，保持发送者和轮数不变：`125ms` 约等于 8 msg/s，`100ms` 约等于 10 msg/s，`83ms` 约等于 12 msg/s，`67ms` 约等于 15 msg/s，`56ms` 约等于 18 msg/s。先跑 `idle`，再从低速 `group` 逐级上探；某一级出现连接波动、ACK 长尾或 DB pool wait 明显上升时，重复该级确认后再进入下一档。 + +自动重连由 `reconnect_workers`、`reconnect_min_wait`、`reconnect_max_wait` 控制。断线账号会按指数退避并附带确定性抖动后重连，避免两台压测机在入口抖动时同时打出登录洪峰。 ## 机器 A：回收并合并报告 diff --git a/loadtest/im-pressure/agent.go b/loadtest/im-pressure/agent.go index e51e36e..4ca0ed6 100644 --- a/loadtest/im-pressure/agent.go +++ b/loadtest/im-pressure/agent.go @@ -49,7 +49,7 @@ type agentRun struct { } func validScenario(scenario string) bool { - return scenario == "p2p" || scenario == "group" || scenario == "mixed" + return scenario == "p2p" || scenario == "group" || scenario == "mixed" || scenario == "idle" } func preflightP2P(ctx context.Context, cfg *Config, manifest *Manifest, local []AccountRecord, apiKey string) error { @@ -680,10 +680,25 @@ func (r *agentRun) connect(ctx context.Context, accounts []AccountRecord, groupT } func (r *agentRun) maintainConnections(ctx context.Context, accounts []AccountRecord, groupTopic, apiKey, password string) { + type reconnectState struct { + attempt int + nextAt time.Time + } byIndex := make(map[int]AccountRecord, len(accounts)) for _, account := range accounts { byIndex[account.Index] = account } + states := make(map[int]reconnectState, len(accounts)) + var stateMu sync.Mutex + minWait := mustDuration(r.cfg.ReconnectMinWait) + maxWait := mustDuration(r.cfg.ReconnectMaxWait) + if maxWait < minWait { + maxWait = minWait + } + reconnectWorkers := r.cfg.ReconnectWorkers + if reconnectWorkers <= 0 { + reconnectWorkers = 8 + } ticker := time.NewTicker(time.Second) defer ticker.Stop() for { @@ -692,7 +707,7 @@ func (r *agentRun) maintainConnections(ctx context.Context, accounts []AccountRe } jobs := make(chan AccountRecord) var wg sync.WaitGroup - workers := 16 + workers := reconnectWorkers if workers > len(accounts) { workers = len(accounts) } @@ -714,8 +729,17 @@ func (r *agentRun) maintainConnections(ctx context.Context, accounts []AccountRe if ctx.Err() == nil { log.Printf("account=%s reconnect failed: %v", account.Username, err) } + stateMu.Lock() + state := states[index] + state.attempt++ + state.nextAt = time.Now().Add(reconnectBackoff(minWait, maxWait, state.attempt, index)) + states[index] = state + stateMu.Unlock() continue } + stateMu.Lock() + delete(states, index) + stateMu.Unlock() r.setClient(index, client) log.Printf("account=%s connection restored", account.Username) } @@ -732,6 +756,12 @@ func (r *agentRun) maintainConnections(ctx context.Context, accounts []AccountRe if current != nil && current.alive() { continue } + stateMu.Lock() + nextAt := states[index].nextAt + stateMu.Unlock() + if !nextAt.IsZero() && time.Now().Before(nextAt) { + continue + } select { case jobs <- account: case <-ctx.Done(): @@ -750,6 +780,25 @@ func (r *agentRun) maintainConnections(ctx context.Context, accounts []AccountRe } } +func reconnectBackoff(minWait, maxWait time.Duration, attempt, index int) time.Duration { + if attempt < 1 { + attempt = 1 + } + delay := minWait + for i := 1; i < attempt && delay < maxWait; i++ { + delay *= 2 + } + if delay > maxWait { + delay = maxWait + } + jitterBase := delay / 4 + if jitterBase <= 0 { + return delay + } + jitter := time.Duration((index*97+attempt*53)%1000) * jitterBase / 1000 + return delay + jitter +} + func (r *agentRun) connectAccountWithRetry(ctx context.Context, account AccountRecord, groupTopic, apiKey, password string) (*WSClient, error) { var lastErr error for attempt := 1; attempt <= 6; attempt++ { @@ -787,12 +836,25 @@ func (r *agentRun) execute(ctx context.Context, start time.Time, groupTopic stri case "p2p": return r.scheduleP2P(ctx, start, r.cfg.UsersPerMachine, r.cfg.P2PMessages, mustDuration(r.cfg.P2PInterval)) case "group": - return r.scheduleGroup(ctx, start, groupTopic, 1, scaledUsers(r.cfg.UsersPerMachine, 1, 2), 2) + senders := r.cfg.GroupSenders + if senders <= 0 { + senders = scaledUsers(r.cfg.UsersPerMachine, 1, 2) + } + if senders > r.cfg.UsersPerMachine { + return fmt.Errorf("group_senders_per_machine=%d exceeds users_per_machine=%d", senders, r.cfg.UsersPerMachine) + } + return r.scheduleGroup(ctx, start, groupTopic, 1, senders, r.cfg.GroupRounds, mustDuration(r.cfg.GroupInterval)) case "mixed": - p2pUsers := scaledUsers(r.cfg.UsersPerMachine, 3, 10) - groupSenders := scaledUsers(r.cfg.UsersPerMachine, 1, 4) + p2pUsers := r.cfg.MixedP2PUsers + if p2pUsers <= 0 { + p2pUsers = scaledUsers(r.cfg.UsersPerMachine, 3, 10) + } + groupSenders := r.cfg.MixedGroupSenders + if groupSenders <= 0 { + groupSenders = scaledUsers(r.cfg.UsersPerMachine, 1, 4) + } if p2pUsers+groupSenders > r.cfg.UsersPerMachine { - groupSenders = r.cfg.UsersPerMachine - p2pUsers + return fmt.Errorf("mixed_p2p_users_per_machine + mixed_group_senders_per_machine exceeds users_per_machine") } if groupSenders <= 0 { return fmt.Errorf("users_per_machine=%d is too small for mixed scenario", r.cfg.UsersPerMachine) @@ -803,7 +865,7 @@ func (r *agentRun) execute(ctx context.Context, start time.Time, groupTopic stri go func() { defer wg.Done(); errCh <- r.scheduleP2P(ctx, start, p2pUsers, 5, 20*time.Second) }() go func() { defer wg.Done() - errCh <- r.scheduleGroup(ctx, start, groupTopic, p2pUsers+1, p2pUsers+groupSenders, 2) + errCh <- r.scheduleGroup(ctx, start, groupTopic, p2pUsers+1, p2pUsers+groupSenders, r.cfg.GroupRounds, mustDuration(r.cfg.GroupInterval)) }() wg.Wait() close(errCh) @@ -813,6 +875,8 @@ func (r *agentRun) execute(ctx context.Context, start time.Time, groupTopic stri } } return nil + case "idle": + return waitContext(ctx, mustDuration(r.cfg.IdleDuration)) default: return fmt.Errorf("unknown scenario %s", r.scenario) } @@ -867,15 +931,19 @@ func (r *agentRun) scheduleP2P(ctx context.Context, start time.Time, users, mess return ctx.Err() } -func (r *agentRun) scheduleGroup(ctx context.Context, start time.Time, topic string, first, last, rounds int) error { +func (r *agentRun) scheduleGroup(ctx context.Context, start time.Time, topic string, first, last, rounds int, globalInterval time.Duration) error { if first < 1 || last > r.clientCount() || first > last { return fmt.Errorf("invalid group publisher range %d-%d", first, last) } count := last - first + 1 - roundDuration := time.Duration(count) * 200 * time.Millisecond + if globalInterval <= 0 { + return fmt.Errorf("group publish interval must be positive") + } + perMachineInterval := 2 * globalInterval + roundDuration := time.Duration(count) * perMachineInterval offset := time.Duration(0) if r.machine == "b" { - offset = 100 * time.Millisecond + offset = globalInterval } var wg sync.WaitGroup for index := first; index <= last; index++ { @@ -884,7 +952,7 @@ func (r *agentRun) scheduleGroup(ctx context.Context, start time.Time, topic str go func(index, position int) { defer wg.Done() for round := 1; round <= rounds; round++ { - target := start.Add(time.Duration(round-1)*roundDuration + time.Duration(position)*200*time.Millisecond + offset) + target := start.Add(time.Duration(round-1)*roundDuration + time.Duration(position)*perMachineInterval + offset) if waitUntil(ctx, target) != nil { return } diff --git a/loadtest/im-pressure/agent_history_test.go b/loadtest/im-pressure/agent_history_test.go index 8b1e024..92911f8 100644 --- a/loadtest/im-pressure/agent_history_test.go +++ b/loadtest/im-pressure/agent_history_test.go @@ -76,6 +76,33 @@ func TestScenarioUsersScaleWithConfiguredCapacity(t *testing.T) { } } +func TestReconnectBackoffCapsAndJitters(t *testing.T) { + minWait := 2 * time.Second + maxWait := 30 * time.Second + first := reconnectBackoff(minWait, maxWait, 1, 10) + if first < minWait || first >= minWait+minWait/4 { + t.Fatalf("first backoff=%s outside expected jitter range", first) + } + capped := reconnectBackoff(minWait, maxWait, 10, 10) + if capped < maxWait || capped >= maxWait+maxWait/4 { + t.Fatalf("capped backoff=%s outside expected jitter range", capped) + } + if reconnectBackoff(minWait, maxWait, 3, 10) == reconnectBackoff(minWait, maxWait, 3, 11) { + t.Fatal("different account indexes should receive different jitter") + } +} + +func TestValidScenarioIncludesIdle(t *testing.T) { + for _, scenario := range []string{"p2p", "group", "mixed", "idle"} { + if !validScenario(scenario) { + t.Fatalf("scenario %s should be valid", scenario) + } + } + if validScenario("unknown") { + t.Fatal("unknown scenario should be rejected") + } +} + func TestNewHistoryQueryIncludesInterleavedSequenceRange(t *testing.T) { messages := make([]AcceptedMessage, 10) for i := range messages { diff --git a/loadtest/im-pressure/config.example.json b/loadtest/im-pressure/config.example.json index 74d372e..d58a094 100644 --- a/loadtest/im-pressure/config.example.json +++ b/loadtest/im-pressure/config.example.json @@ -13,6 +13,9 @@ "insecure_tls": false, "prepare_workers": 20, "connect_ramp": "20s", + "reconnect_workers": 8, + "reconnect_min_wait": "2s", + "reconnect_max_wait": "30s", "command_timeout": "20s", "ws_ping_interval": "20s", "preflight_workers": 32, @@ -26,6 +29,12 @@ "close_ramp": "10s", "p2p_messages_per_user": 10, "p2p_interval": "1s", + "group_senders_per_machine": 500, + "group_rounds": 2, + "group_publish_interval": "100ms", + "mixed_p2p_users_per_machine": 300, + "mixed_group_senders_per_machine": 250, + "idle_duration": "15m", "drain_p2p": "30s", "drain_group": "60s", "delivery_quiet_period": "10s", diff --git a/loadtest/im-pressure/config.single-machine.example.json b/loadtest/im-pressure/config.single-machine.example.json index 330ac37..5629cec 100644 --- a/loadtest/im-pressure/config.single-machine.example.json +++ b/loadtest/im-pressure/config.single-machine.example.json @@ -17,6 +17,9 @@ "insecure_tls": false, "prepare_workers": 20, "connect_ramp": "30s", + "reconnect_workers": 8, + "reconnect_min_wait": "2s", + "reconnect_max_wait": "30s", "command_timeout": "90s", "ws_ping_interval": "20s", "preflight_workers": 32, @@ -30,6 +33,12 @@ "close_ramp": "60s", "p2p_messages_per_user": 10, "p2p_interval": "1s", + "group_senders_per_machine": 300, + "group_rounds": 2, + "group_publish_interval": "100ms", + "mixed_p2p_users_per_machine": 180, + "mixed_group_senders_per_machine": 150, + "idle_duration": "15m", "drain_p2p": "30s", "drain_group": "60s", "delivery_quiet_period": "10s", diff --git a/loadtest/im-pressure/main.go b/loadtest/im-pressure/main.go index db7d5d3..e7ff97e 100644 --- a/loadtest/im-pressure/main.go +++ b/loadtest/im-pressure/main.go @@ -101,7 +101,7 @@ func coordinatorCommand(ctx context.Context, args []string) error { fs := flag.NewFlagSet("coordinator", flag.ContinueOnError) listen := fs.String("listen", ":8090", "coordinator listen address") runID := fs.String("run-id", "", "unique run ID, max 24 characters") - scenario := fs.String("scenario", "", "p2p, group, or mixed") + scenario := fs.String("scenario", "", "p2p, group, mixed, or idle") output := fs.String("output", "loadtest/im-pressure/results/coordinator.json", "coordinator result path") expected := fs.Int("expected-agents", 2, "number of agents") delay := fs.Duration("start-delay", 30*time.Second, "delay after all agents are ready") diff --git a/loadtest/im-pressure/model.go b/loadtest/im-pressure/model.go index 85b1441..52b1efb 100644 --- a/loadtest/im-pressure/model.go +++ b/loadtest/im-pressure/model.go @@ -16,35 +16,44 @@ import ( const manifestVersion = 1 type Config struct { - Tenant string `json:"tenant"` - HTTPBase string `json:"http_base"` - APIKeyEnv string `json:"api_key_env"` - PasswordEnv string `json:"password_env"` - UsersPerMachine int `json:"users_per_machine"` - Endpoints map[string][]string `json:"endpoints"` - DialIPs map[string][]string `json:"dial_ips,omitempty"` - InsecureTLS bool `json:"insecure_tls"` - PrepareWorkers int `json:"prepare_workers"` - ConnectRamp string `json:"connect_ramp"` - CommandTimeout string `json:"command_timeout"` - WSPingInterval string `json:"ws_ping_interval"` - P2PMessages int `json:"p2p_messages_per_user"` - P2PInterval string `json:"p2p_interval"` - DrainP2P string `json:"drain_p2p"` - DrainGroup string `json:"drain_group"` - DeliveryQuiet string `json:"delivery_quiet_period"` - DeliveryTimeout string `json:"delivery_quiet_timeout"` - PhonePrefix string `json:"registration_phone_prefix"` - VerifyCodeEnv string `json:"verification_code_env"` - HistoryWorkers int `json:"history_workers"` - HistoryPageSize int `json:"history_page_size"` - HistoryInterval string `json:"history_query_interval"` - HistoryPageWait string `json:"history_page_wait"` - HistorySettle string `json:"history_settle"` - HistoryTimeout string `json:"history_timeout"` - CloseRamp string `json:"close_ramp"` - PreflightWorkers int `json:"preflight_workers"` - SkipP2PPreflight bool `json:"skip_p2p_preflight"` + Tenant string `json:"tenant"` + HTTPBase string `json:"http_base"` + APIKeyEnv string `json:"api_key_env"` + PasswordEnv string `json:"password_env"` + UsersPerMachine int `json:"users_per_machine"` + Endpoints map[string][]string `json:"endpoints"` + DialIPs map[string][]string `json:"dial_ips,omitempty"` + InsecureTLS bool `json:"insecure_tls"` + PrepareWorkers int `json:"prepare_workers"` + ConnectRamp string `json:"connect_ramp"` + ReconnectWorkers int `json:"reconnect_workers"` + ReconnectMinWait string `json:"reconnect_min_wait"` + ReconnectMaxWait string `json:"reconnect_max_wait"` + CommandTimeout string `json:"command_timeout"` + WSPingInterval string `json:"ws_ping_interval"` + P2PMessages int `json:"p2p_messages_per_user"` + P2PInterval string `json:"p2p_interval"` + GroupSenders int `json:"group_senders_per_machine"` + GroupRounds int `json:"group_rounds"` + GroupInterval string `json:"group_publish_interval"` + MixedP2PUsers int `json:"mixed_p2p_users_per_machine"` + MixedGroupSenders int `json:"mixed_group_senders_per_machine"` + IdleDuration string `json:"idle_duration"` + DrainP2P string `json:"drain_p2p"` + DrainGroup string `json:"drain_group"` + DeliveryQuiet string `json:"delivery_quiet_period"` + DeliveryTimeout string `json:"delivery_quiet_timeout"` + PhonePrefix string `json:"registration_phone_prefix"` + VerifyCodeEnv string `json:"verification_code_env"` + HistoryWorkers int `json:"history_workers"` + HistoryPageSize int `json:"history_page_size"` + HistoryInterval string `json:"history_query_interval"` + HistoryPageWait string `json:"history_page_wait"` + HistorySettle string `json:"history_settle"` + HistoryTimeout string `json:"history_timeout"` + CloseRamp string `json:"close_ramp"` + PreflightWorkers int `json:"preflight_workers"` + SkipP2PPreflight bool `json:"skip_p2p_preflight"` } func loadConfig(path string) (*Config, error) { @@ -68,6 +77,29 @@ func loadConfig(path string) (*Config, error) { if cfg.ConnectRamp == "" { cfg.ConnectRamp = "20s" } + if cfg.ReconnectWorkers <= 0 { + cfg.ReconnectWorkers = 8 + } + if cfg.ReconnectWorkers > 128 { + return nil, errors.New("reconnect_workers must not exceed 128") + } + if cfg.ReconnectMinWait == "" { + cfg.ReconnectMinWait = "2s" + } + if _, err = configDuration(cfg.ReconnectMinWait); err != nil { + return nil, fmt.Errorf("reconnect_min_wait: %w", err) + } + if cfg.ReconnectMaxWait == "" { + cfg.ReconnectMaxWait = "30s" + } + minReconnect, _ := configDuration(cfg.ReconnectMinWait) + maxReconnect, err := configDuration(cfg.ReconnectMaxWait) + if err != nil { + return nil, fmt.Errorf("reconnect_max_wait: %w", err) + } + if maxReconnect < minReconnect { + return nil, errors.New("reconnect_max_wait must be greater than or equal to reconnect_min_wait") + } if cfg.CommandTimeout == "" { cfg.CommandTimeout = "20s" } @@ -83,6 +115,30 @@ func loadConfig(path string) (*Config, error) { if cfg.P2PInterval == "" { cfg.P2PInterval = "1s" } + if cfg.GroupRounds <= 0 { + cfg.GroupRounds = 2 + } + if cfg.GroupInterval == "" { + cfg.GroupInterval = "100ms" + } + if _, err = configDuration(cfg.GroupInterval); err != nil { + return nil, fmt.Errorf("group_publish_interval: %w", err) + } + if cfg.GroupSenders < 0 { + return nil, errors.New("group_senders_per_machine must not be negative") + } + if cfg.MixedP2PUsers < 0 { + return nil, errors.New("mixed_p2p_users_per_machine must not be negative") + } + if cfg.MixedGroupSenders < 0 { + return nil, errors.New("mixed_group_senders_per_machine must not be negative") + } + if cfg.IdleDuration == "" { + cfg.IdleDuration = "15m" + } + if _, err = configDuration(cfg.IdleDuration); err != nil { + return nil, fmt.Errorf("idle_duration: %w", err) + } if cfg.DrainP2P == "" { cfg.DrainP2P = "30s" }