From 2b90a2e75937f500995c56fa8028e4bfd7a2ec2d Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Sat, 26 Sep 2026 21:50:43 +0800 Subject: [PATCH] feat: add dual-host IM pressure test tool --- docs/IM-006商户双机消息压力测试方案.md | 792 +++++++++++++++++++++++ docs/IM-多开跨Pod实时投递压测方案.md | 3 + loadtest/im-pressure/.gitignore | 7 + loadtest/im-pressure/README.md | 123 ++++ loadtest/im-pressure/agent.go | 461 +++++++++++++ loadtest/im-pressure/agent_metrics.go | 68 ++ loadtest/im-pressure/collector.go | 114 ++++ loadtest/im-pressure/config.example.json | 19 + loadtest/im-pressure/coordinator.go | 323 +++++++++ loadtest/im-pressure/coordinator_test.go | 60 ++ loadtest/im-pressure/histogram.go | 64 ++ loadtest/im-pressure/httpapi.go | 259 ++++++++ loadtest/im-pressure/main.go | 146 +++++ loadtest/im-pressure/merge.go | 114 ++++ loadtest/im-pressure/merge_test.go | 49 ++ loadtest/im-pressure/model.go | 294 +++++++++ loadtest/im-pressure/model_test.go | 45 ++ loadtest/im-pressure/prepare.go | 253 ++++++++ loadtest/im-pressure/recorder.go | 229 +++++++ loadtest/im-pressure/recorder_test.go | 55 ++ loadtest/im-pressure/wsclient.go | 365 +++++++++++ 21 files changed, 3843 insertions(+) create mode 100644 docs/IM-006商户双机消息压力测试方案.md create mode 100644 loadtest/im-pressure/.gitignore create mode 100644 loadtest/im-pressure/README.md create mode 100644 loadtest/im-pressure/agent.go create mode 100644 loadtest/im-pressure/agent_metrics.go create mode 100644 loadtest/im-pressure/collector.go create mode 100644 loadtest/im-pressure/config.example.json create mode 100644 loadtest/im-pressure/coordinator.go create mode 100644 loadtest/im-pressure/coordinator_test.go create mode 100644 loadtest/im-pressure/histogram.go create mode 100644 loadtest/im-pressure/httpapi.go create mode 100644 loadtest/im-pressure/main.go create mode 100644 loadtest/im-pressure/merge.go create mode 100644 loadtest/im-pressure/merge_test.go create mode 100644 loadtest/im-pressure/model.go create mode 100644 loadtest/im-pressure/model_test.go create mode 100644 loadtest/im-pressure/prepare.go create mode 100644 loadtest/im-pressure/recorder.go create mode 100644 loadtest/im-pressure/recorder_test.go create mode 100644 loadtest/im-pressure/wsclient.go diff --git a/docs/IM-006商户双机消息压力测试方案.md b/docs/IM-006商户双机消息压力测试方案.md new file mode 100644 index 0000000..e798d83 --- /dev/null +++ b/docs/IM-006商户双机消息压力测试方案.md @@ -0,0 +1,792 @@ +# IM 006 商户双机消息压力测试方案 + +日期：2026-09-25 + +## 1. 测试目标 + +本次测试使用 006 商户的 2000 个账号，在两台压测机上各运行 1000 个在线 WebSocket 用户，验证多 Pod 场景下： + +1. 1000 对 P2P 会话同时收发时，消息是否被接受、实时投递并最终持久化。 +2. 2000 人大群中 1000 个成员轮流发言时，跨 Pod 群广播的丢失率、重复率、乱序率和延迟。 +3. 300 对 P2P 与 500 个群发送者同时工作时，混合负载是否相互影响。 +4. Tinode 原生 Cluster、Kafka 消息持久化、数据库分表和 WebSocket 下行是否出现容量瓶颈。 + +本轮不在正式发压窗口内注册账号、加好友或建群。所有数据准备完成并核验后，再开始消息压力测试。 + +## 2. 本方案对需求的精确定义 + +### 2.1 账号与机器分片 + +账号使用只包含小写字母和数字、且不超过 10 个字符的用户名，以满足当前注册校验： + +| 压测机 | 账号 | 数量 | 用户名示例 | +| --- | --- | ---: | --- | +| Machine A | A0001-A1000 | 1000 | `lta0001`、`lta1000` | +| Machine B | B0001-B1000 | 1000 | `ltb0001`、`ltb1000` | + +固定配对关系为 `A0001 <-> B0001`，依此类推到 `A1000 <-> B1000`。“号码相同”指账号序号相同，不使用相同手机号。 + +每个账号最终记录：`username`、`user_id`、`token`、`token_expires`、`machine`、`pair_no`、`p2p_topic`、`group_topic`。密码只从环境变量读取，不写入结果文件或日志。 + +### 2.2 群聊需求解释 + +“1000 个人间隔 0.1 秒发消息，2 轮合计 2000 条”按全局调度解释： + +- 每 100ms 只安排一个发送者发送一条消息。 +- 每轮 1000 个发送者各发送 1 条，共 1000 条，持续约 100 秒。 +- 连续执行 2 轮，共 2000 条，正式发压约 200 秒。 +- 不是每个发送者每 100ms 发一条；后一种解释会产生 10000 条/秒和 100 万条/100 秒，与“合计 2000 条”矛盾。 + +### 2.3 P2P 场景默认强度 + +原需求未指定每对消息数。本方案默认每对双方各发 10 条： + +- 1000 对同时开始。 +- 每个用户每秒发送 1 条，共 10 秒；每次加 0-200ms 确定性抖动，避免所有请求落在同一毫秒。 +- 总发布量：`1000 × 2 × 10 = 20000` 条。 +- 主要期望投递量：20000 次对端投递；发送方自身回显单独统计，不混入对端丢失率。 + +消息数、间隔和抖动必须可通过命令行参数修改，但每轮结果必须记录实际参数。 + +## 3. 协议与接口契约 + +### 3.1 WebSocket 建连与账号注册 + +WebSocket 地址为 `/v0/channels`，Upgrade 携带普通 API Key。每个连接严格执行： + +```text +Upgrade +-> hi(tenant="006") +-> acc 或 login +-> sub(me) +-> sub(目标 P2P/群 Topic) +-> READY +``` + +006 是企业商户，注册使用 `registration_type: 0` 和 `tenant_code: "006"`： + +```json +{ + "acc": { + "id": "prep-acc-lta0001", + "user": "new", + "scheme": "basic", + "username": "lta0001", + "registration_type": 0, + "tenant_code": "006", + "secret": "base64(lta0001:password)", + "login": true, + "desc": {"public": {"fn": "LT A0001"}} + } +} +``` + +当前 006 环境必须确认未强制手机号或邀请码注册。准备程序应先尝试 basic 登录：已存在且密码正确则复用；仅在账号不存在时注册。账号存在但密码不匹配必须停止，禁止覆盖真实账号。 + +### 3.2 HTTP 公共请求头 + +好友、隐私和群接口使用登录返回的同一 Token： + +```text +X-MeetingIM-APIKey: +X-IM-Schema: 1 +X-Request-ID: +Authorization: Bearer +Idempotency-Key: +``` + +所有密钥和 Token 通过环境变量或权限受控文件传入，日志只记录账号索引和脱敏 ID。 + +### 3.3 好友准备 + +对每个序号执行： + +1. A 调用 `POST /v1/im/friend-requests`，Body 为 `{"target_user_id":"usr_b","message":"loadtest","source":"search","source_topic":""}`。 +2. 若返回 `pending`，B 调用 `POST /v1/im/friend-requests/{request_id}/accept`。 +3. 若已经是好友，读取好友列表或会话 context 并复用关系。 +4. 保存响应中的 P2P `topic`；必要时调用 `POST /v1/im/conversations/p2p`，Body 为 `{"peer_user_id":"usr_b"}`。 +5. 最终核验 1000 对关系均为 `friend`，且双方能订阅同一个 P2P Topic。 + +好友创建会产生 `friend_added` 系统消息，因此正式统计只识别本轮 `run_id` 对应的 `x-im-client-mid`，不能把 Topic 的初始 `seq` 当成 0。 + +### 3.4 2000 人群准备 + +1. 使用 A0001 调用 `POST /v1/im/groups` 创建群。创建接口最多附带 99 个成员，建议只创建群主。 +2. 先确认所有账号的 `group_invite_verification=false`，避免进入当前尚未完整开放的邀请确认链路。 +3. 调用 `POST /v1/im/groups/{topic}/invitations` 分批邀请，每批不超过 100 人，每批使用独立 `Idempotency-Key`。 +4. 邀请完成后分页调用 `GET /v1/im/groups/{topic}/members`，核验去重后的有效成员恰好为 2000。 +5. 2000 个 WebSocket 全部订阅该群，并完成屏障消息验证后才允许发压。 + +直接邀请会产生群业务事件和系统消息。准备结束后至少静默 30 秒，并记录正式测试前的群 `baseline_seq`；统计时仍以 `run_id/client_mid` 为准。 + +## 4. 环境硬门槛 + +任一门槛不满足，本轮标记为 `INVALID`，不能把结果当作服务端性能结论： + +1. `max_subscriber_count >= 2500`。仓库配置默认是 128，必须在所有 Pod 上统一调高并滚动重启。 +2. 006 商户 `max_group_members` 为 0（不限）或不小于 2000；目标群 `max_member_count` 同样满足要求。 +3. IM Server 所有 Pod 版本和配置一致，原生 Cluster 成员数与 Running Pod 数一致。 +4. Kafka 的 P2P、群消息持久化 Topic 可写，消费组无持续积压，死信在基线时为 0。 +5. 两台压测机到目标入口网络稳定，文件描述符上限、临时端口和 NAT 容量足够。 +6. 两台压测机使用 chrony/NTP，同一时刻的时钟偏差绝对值小于 5ms，否则跨机器单向延迟无效。 +7. 每台压测机至少预留 2 CPU、4GB 内存和稳定的 100Mbps 网络；测试进程自身 CPU 不应长期超过 70%。 +8. 2000 个连接全部 READY；正式阶段不允许把连接失败用户从分母中静默删除。 + +## 5. 多 Pod 拓扑 + +仅通过同一个负载均衡入口连接，不能证明消息确实跨 Pod。推荐提供仅供压测使用的三个直连入口，并保持业务数据路径不变： + +```text +Machine A: A0001-A0500 -> Pod 0 + A0501-A1000 -> Pod 1 +Machine B: B0001-B1000 -> Pod 2 +``` + +这样 1000 对 P2P 的双方一定不在同一个 Pod；群成员也覆盖全部 3 个 Pod。若只能提供两个直连入口，则 Machine A 固定 Pod 0、Machine B 固定 Pod 1，并在另一轮把 Machine B 切到 Pod 2 重跑。 + +入口必须能明确确认实际 Pod，例如握手诊断日志或反向代理访问日志能关联 `session -> pod`。不要依赖负载均衡概率分布。 + +Topic 消息在启用 Tinode 原生 Cluster 时由 Cluster 路由；此时 Redis realtime 指标中的 `publish/skip_native_cluster` 是预期状态，不代表丢失。Redis 仍需观察用户级业务事件和 Session 控制类 fanout。 + +## 6. 统一消息格式 + +所有正式测试消息都使用 JSON 文本消息，并设置 `noecho:false`： + +```json +{ + "pub": { + "id": "s1-a0001-00001", + "topic": "p2p_or_grp", + "noecho": false, + "head": { + "mime": "application/json", + "x-im-schema": 1, + "x-im-type": "text", + "x-im-client-mid": "lt-20260925-s1-a0001-00001" + }, + "content": { + "text": "loadtest", + "run_id": "lt-20260925-s1", + "scenario": "p2p", + "sender_index": 1, + "message_index": 1, + "sent_at_unix_nano": 0 + } + } +} +``` + +`x-im-client-mid` 在 `tenant + topic + sender` 范围内必须唯一，重试时必须复用原值。发送程序用 `{ctrl.id}` 关联请求，用 `ctrl.params.seq` 记录服务端分配的 Topic 序号。 + +## 7. 执行阶段 + +### 7.1 阶段 P0：预检与热身 + +1. 两台机器各以 50 连接/秒建立 1000 条连接，总建连时间约 20 秒。 +2. 每个用户订阅 `me`、自己的 P2P Topic 和群 Topic。 +3. 群主发送 1 条不计入正式结果的群屏障消息；必须在 10 秒内被 2000 个连接全部收到。 +4. 每个 P2P Topic 发送 1 条不计入正式结果的屏障消息；双方必须收到，且对端投递为 1000/1000。 +5. 等待 30 秒，确认错误计数、Kafka lag、CPU、内存和连接数回到稳定状态。 + +### 7.2 阶段 S1：1000 对 P2P 同时聊天 + +参与者：全部 2000 人，1000 个 P2P Topic。 + +默认调度： + +- A、B 双方各发送 10 条，每用户 1 条/秒，持续 10 秒。 +- 每个用户使用由账号索引计算的 0-200ms 固定抖动，确保可复现。 +- 总发布量 20000，平均发布速率约 2000 条/秒。 +- 发布完成后保留连接并等待 30 秒排空，再进行历史核对。 + +主要期望：每条被服务端接受的消息恰好被对应的另一方实时收到一次。发送方回显也应收到一次，但作为独立指标。 + +### 7.3 阶段 S2：2000 人群、1000 人发言 + +参与者：2000 个在线订阅者。发送者选择 A0001-A0500 与 B0001-B0500，共 1000 人，避免所有发布压力只落在一台压测机上。 + +默认调度： + +- Machine A 每 200ms 发 1 条，Machine B 在其后错开 100ms 发 1 条，合并后全局间隔为 100ms。 +- 每轮每个发送者恰好发送 1 条，共 1000 条、约 100 秒。 +- 连续 2 轮，共 2000 条、约 200 秒。 +- `noecho:false`，因此每条被接受的消息期望投递到 2000 个在线连接。 +- 理论实时投递总量：`2000 × 2000 = 4000000` 次，平均约 20000 次/秒。 +- 发布结束后等待 60 秒排空，然后由多个校验账号分页拉取历史。 + +### 7.4 阶段 S3：P2P 与群聊混合 + +正式窗口为 100 秒： + +- P2P：A0001-A0300 与 B0001-B0300，共 300 对；双方各发 5 条，总发布量 `300 × 2 × 5 = 3000`。在第 0、20、40、60、80 秒附近发送，并加入固定抖动。 +- 群聊：A0301-A0550 与 B0301-B0550，共 500 个发送者；每人发 2 条。两台机器错峰形成全局 100ms 间隔，两轮共 1000 条，持续约 100 秒。 +- 其余用户保持在线并接收群消息。 +- 群实时理论投递总量：`1000 × 2000 = 2000000` 次。 +- P2P 主要理论对端投递总量：3000 次；发送方回显另计 3000 次。 +- 发布结束后等待 60 秒排空，再做 P2P 和群历史核对。 + +混合场景的群发送者与 P2P 发送者故意分开，便于判断瓶颈来自连接写队列、群广播还是同一连接并发写。通过后可增加一轮“发送者重叠”增强测试。 + +## 8. 丢失率与正确性口径 + +不能只统计压测端成功调用 `WriteJSON` 的数量。每轮必须同时输出以下四层结果。 + +### 8.1 发送接受率 + +```text +publish_accept_rate = ctrl_202_with_seq / publish_attempted +publish_reject_rate = (ctrl_4xx + ctrl_5xx + ctrl_timeout + write_error) / publish_attempted +``` + +只有收到 `ctrl.code=202` 且包含有效 `params.seq` 的消息，才进入后续“期望投递”分母。 + +### 8.2 实时投递丢失率 + +实时到达率必须由 2000 个 WebSocket 连接实际监听得到，不能使用发送 ACK、Kafka offset 或数据库行数代替。每个连接建立后启动一个常驻 read loop，全量读取服务端下行，并把 `{ctrl}`、`{data}`、`{meta}`、`{pres}`、`{info}` 分流处理。 + +监听器要求： + +1. 每个 WebSocket 只能有一个 goroutine 调用 `ReadMessage/ReadJSON`，避免多个读取者争抢数据；`ctrl` 通过 `id` 分发给等待中的发布请求。 +2. 只统计 `head.x-im-client-mid` 前缀和 `content.run_id` 都匹配本轮的 `{data}`，忽略准备阶段系统消息、其他业务消息和历史噪声。 +3. 记录 `receiver_index + client_mid + seq + first_received_at`。相同接收者再次收到同一 `client_mid` 计为重复，不增加到达数。 +4. 服务端允许 `{data}` 早于对应 `{ctrl 202}` 到达，因此监听器先保存观测事实，最终再与 ACK ledger 对账，不能因为 ACK 尚未到达而丢弃该 `{data}`。 +5. P2P 以配对的另一方为主要期望接收者；发送者自身回显单独统计。 +6. 群消息以发压开始时的 2000 个 READY 成员为固定分母。因为 `noecho:false`，发送者本人也属于期望接收者。 +7. 正式窗口内连接断开时不得从分母中删除。记录断开时间、所属机器和 Pod；服务端导致的断开计入结果，压测机自身饱和导致的断开则将整轮标记为无效。 +8. P2P 发布结束后继续监听 30 秒，群聊和混合场景继续监听 60 秒。窗口内未收到计为实时丢失；窗口后到达另记 `late_delivery`，不能把它改写为正常实时到达。 + +期望投递账本由成功 ACK 生成： + +```text +P2P expected key = run_id + client_mid + peer_receiver_index +Group expected key = run_id + client_mid + each_ready_receiver_index +``` + +群场景 2000 条消息对应 400 万个期望键。实现时为每个接收者维护按消息序号索引的 bitset，不保存 400 万份完整 JSON；延迟使用 HDR Histogram 或固定桶，缺失和重复才输出逐条明细。 + +P2P 主要口径： + +```text +p2p_realtime_loss = + (accepted_messages - unique_peer_data_received_within_drain_window) + / accepted_messages +``` + +群聊口径： + +```text +group_expected_deliveries = accepted_group_messages × ready_group_subscribers +group_realtime_loss = + (group_expected_deliveries - unique_group_deliveries_within_drain_window) + / group_expected_deliveries +``` + +每个接收者按 `run_id + x-im-client-mid` 去重。重复到达不用于抵消其他消息的缺失。 + +同时输出： + +```text +arrival_rate = unique_deliveries_within_drain_window / expected_deliveries +loss_rate = 1 - arrival_rate +late_rate = unique_late_deliveries / expected_deliveries +duplicate_rate = duplicate_deliveries / expected_deliveries +``` + +实时监听结束后继续执行历史回查，将缺失分为： + +- `realtime_lost_but_persisted`：WS 窗口内没收到，但历史中存在，说明实时投递链路丢失，客户端可通过重连补偿。 +- `persisted_but_late`：超过实时排空窗口才由 WS 收到，同时历史中存在。 +- `durable_lost`：WS 未收到且最长等待后历史也不存在，属于真正的持久化消息丢失。 +- `receiver_disconnected`：期望接收者在消息窗口内断线，单独统计并保留在实时丢失分母中。 + +### 8.3 持久化丢失率 + +Kafka 持久化是异步链路，`ctrl 202` 不等于数据库已经可查。排空后通过 `{get what:"data"}` 分页回查： + +```text +persistence_loss = + (accepted_unique_messages - unique_messages_found_in_history) + / accepted_unique_messages +``` + +- P2P：1000 个 Topic 各由一侧账号回查。 +- 群聊：至少使用分别连接不同 Pod 的 3 个成员回查，三份结果应完全一致。 +- 最长等待 120 秒，每 5 秒重试一次；超时仍不存在才记为持久化丢失。 + +### 8.4 重复与乱序 + +```text +duplicate_rate = duplicate_deliveries / expected_deliveries +out_of_order_rate = receivers_observing_seq_regression / ready_receivers +``` + +同时记录：同一 `client_mid` 对应多个 `seq`、同一 `seq` 对应多个测试消息、接收端 Topic `seq` 回退、ACK `seq` 与下行 `seq` 不一致。这些都必须单独报错，不能归入普通丢失。 + +## 9. 延迟口径 + +记录以下时间点： + +- `t_write`：压测端开始写 `{pub}`。 +- `t_ack`：收到对应 `{ctrl 202}`。 +- `t_data`：接收者收到并完成 JSON 解析。 + +输出： + +```text +ack_latency = t_ack - t_write +e2e_realtime_latency = t_data - sent_at +``` + +每个场景输出 p50、p95、p99、p99.9 和 max。跨机器延迟仅在 NTP 偏差达标时有效；否则仍可统计 ACK 延迟，但端到端单向延迟标记为无效。 + +群聊 400 万次投递不应把所有延迟样本常驻内存。使用固定桶直方图或 HDR Histogram，并按接收端机器和 Pod 分组。 + +## 10. 建议通过标准 + +第一轮基线建议使用： + +| 指标 | 标准 | +| --- | --- | +| READY 连接 | 2000/2000 | +| 发布接受率 | 100% | +| 持久化丢失率 | 0 | +| 同一 `client_mid` 多 `seq` | 0 | +| P2P 对端实时丢失率 | <= 0.001%，且任何丢失都要输出明细 | +| 群实时投递丢失率 | <= 0.001%，发布版本验收目标为 0 | +| 重复投递率 | 0 | +| Topic `seq` 回退 | 0 | +| ACK p95 / p99 | < 100ms / < 300ms | +| 实时投递 p95 / p99 | < 200ms / < 500ms | +| `deliver/hub_full`、WS 写队列满 | 0 | +| Kafka producer/commit 错误 | 0 | +| Kafka 消费重试、单条降级、死信新增 | 0 | +| Kafka 排空 | 发布结束后 60 秒内 lag 回到基线/0 | +| Kafka 到 MySQL 落库 p95 / p99 | < 200ms / < 500ms | +| MySQL 实际落库数 | 与 `ctrl 202` 唯一消息数完全一致 | +| MySQL deadlock、lock timeout、连接池耗尽 | 0 | +| Redis 操作错误、淘汰、拒绝连接 | 0 | +| Redis command p95 / p99 | < 5ms / < 10ms | +| Pod CPU throttling ratio | < 10% | +| Pod CPU、内存峰值 | < limit 的 80% | +| 容器/节点网络 drop、error | 0 新增或不高于基线噪声 | +| 压测机 CPU、内存、网络 | < 70%，无 read-loop backlog | +| Pod 重启、OOM、持续 GC/内存增长 | 0 | + +如果实时有缺失但历史回查完整，结论应写“实时链路丢失、可由历史恢复”，不能写成“消息未丢失”。如果历史也缺失，则属于持久化消息丢失，直接判失败。 + +## 11. 全链路观测与资源采集 + +### 11.1 观测目标与时间窗口 + +监控必须能把同一轮测试拆成以下链路： + +```text +压测机 WriteJSON +-> IM Pod 收到 pub +-> Redis 分配消息 seq / 幂等 claim +-> Kafka producer 写入消息持久化 Topic +-> IM Pod 返回 ctrl 202 并实时广播 +-> Kafka consumer 拉取与组批 +-> MySQL 批量提交 +-> Kafka offset commit +-> WebSocket 历史查询可见 +``` + +每轮采集窗口为：正式 `start_at` 前 10 分钟、正式测试全程、排空与历史核对结束后 10 分钟。Prometheus 在压测期间建议使用 5 秒抓取周期，重要 counter 以 `increase` 计算，不使用瞬时值推算总量。 + +每轮在 Grafana 或结果文件中记录四个时间点： + +- `baseline_start`：基线开始。 +- `load_start`：正式发送开始。 +- `load_end`：最后一次发布结束。 +- `reconcile_end`：历史核对完成。 + +所有图表使用相同 UTC 时间轴，并标注 `run_id`、scenario、Git commit、Pod 镜像版本和压测参数。 + +### 11.2 采集组件 + +推荐最小采集栈： + +| 对象 | 采集方式 | 主要用途 | +| --- | --- | --- | +| IM Server | `/metrics`、`/debug/vars` | WS、HTTP、DB、fanout、Go runtime、连接池 | +| Kubernetes Pod/Container | kube-state-metrics + kubelet/cAdvisor | CPU、内存、磁盘 IO、网络、重启、限流 | +| Kubernetes Node | node-exporter | 节点 CPU、内存、磁盘、网络、conntrack | +| Kafka | kafka-exporter + Broker JMX Exporter | consumer lag、吞吐、ISR、请求队列、磁盘 | +| Redis | redis-exporter | OPS、延迟、内存、连接、命中率、淘汰、网络 | +| MySQL | mysqld-exporter + performance_schema | QPS、事务、锁、Buffer Pool、慢 SQL、IO | +| 两台压测机 | node-exporter + 压测 agent `/metrics` | 生成端 CPU/网络/连接和客户端统计 | +| 链路探测 | blackbox-exporter 或压测 agent 主动探测 | DNS、TCP、HTTP、依赖 RTT | + +如果环境没有完整 Prometheus 栈，压测前至少以 5 秒周期抓取上述 HTTP 指标，并同步保存 `kubectl top`、Pod 日志、Kafka consumer group、Redis INFO 和 MySQL status。手工快照只能作为兜底，不能替代连续时间序列。 + +### 11.3 当前已有与必须补充的应用指标 + +当前 IM Server 已经提供： + +| 指标 | 状态 | 用途 | +| --- | --- | --- | +| `ws_connections_active` | 已有 | 当前 Pod/实例的活跃 WS 数 | +| `http_requests_total` | 已有 | HTTP 请求量和错误码 | +| `http_request_duration_seconds` | 已有 | HTTP 延迟 | +| `db_query_duration_seconds` | 已有 | MySQL query/exec/commit 等通用耗时 | +| `realtime_fanout_events_total` | 已有 | Redis realtime 发布、接收和本地投递结果 | +| `realtime_fanout_delivery_duration_seconds` | 已有 | Redis realtime 消息年龄 | +| `/debug/vars` 中 `DbStats` | 已有 | Go SQL 连接池状态 | +| `/debug/vars` 中 Cluster/Session/Topic 计数 | 已有 | Cluster 节点、会话和活跃 Topic | + +为准确测量 Kafka 落库速度，压测版本必须再补以下低基数指标。严禁把 `topic`、`user_id`、`client_mid`、`run_id` 或 tenant ID 放入 Prometheus label： + +```text +message_persist_events_total{stage,kind,status} +message_persist_duration_seconds{stage,kind,status} +message_persist_end_to_end_seconds{kind,status} +message_persist_batch_size{kind} +message_persist_inflight{kind} +kafka_consumer_commit_total{topic_kind,status} +redis_operation_duration_seconds{operation,status} +redis_operation_errors_total{operation} +mysql_pool_connections{state} +mysql_pool_wait_total +mysql_pool_wait_duration_seconds_total +cluster_route_events_total{operation,status} +cluster_route_duration_seconds{operation,status} +ws_outbound_events_total{status} +ws_outbound_queue_depth +``` + +标签约定： + +- `kind`：仅允许 `p2p`、`group`。 +- `stage`：`producer_write`、`consumer_fetch`、`db_batch`、`consume_finish`。 +- `status`：`ok`、`error`、`duplicate`、`retry`、`dead`。 +- `operation`：Redis 使用 `seq_reserve`、`unique_claim`、`hot_cache`、`fanout_publish`、`fanout_receive`；Cluster 使用 `route`、`proxy_join`、`proxy_leave`。 + +`message_persist_end_to_end_seconds` 从 Kafka envelope 的 `occurred_at` 计算到 MySQL 提交并标记消费成功的时刻。它是回答“消息从服务端接受到真正落库用了多久”的核心指标。 + +另外应在自定义 Prometheus Registry 中注册 Go collector 和 process collector，提供标准 `go_*`、`process_*` 指标；在完成前从 `/debug/vars` 采集对应 runtime 数据。 + +### 11.4 Kafka 处理速度与积压 + +Kafka 需要同时观察应用层和 Broker 层，至少包含： + +1. P2P Topic `im.message.persist` 的生产/消费条数与字节数。 +2. 群 Topic `im.group.message.persist` 的生产/消费条数与字节数。 +3. 对应 consumer group 每个 partition 的 lag、总 lag 和 lag 恢复到 0 的时间。 +4. producer 写入延迟、错误、重试和 timeout。 +5. consumer fetch 速率、批量大小、DB 批次耗时、offset commit 延迟和失败。 +6. 批量落库失败后降级为单条写入的次数。 +7. 消费重试、Redis 幂等 claim 失败和死信新增数。 +8. Broker request queue、network processor idle、request handler idle、ISR shrink、under-replicated/offline partition 和磁盘使用率。 + +不同 exporter 的 lag 指标名可能是 `kafka_consumergroup_lag` 或 `kafka_consumergroup_group_lag`，部署时必须以实际 `/metrics` 为准建立 dashboard，不允许直接照抄不存在的指标名。 + +典型查询： + +```promql +sum by (kind, stage, status) ( + rate(message_persist_events_total[1m]) +) +``` + +```promql +histogram_quantile( + 0.99, + sum by (le, kind) ( + rate(message_persist_end_to_end_seconds_bucket[5m]) + ) +) +``` + +```promql +sum by (consumergroup, topic) ( + kafka_consumergroup_lag{ + topic=~"im.message.persist|im.group.message.persist" + } +) +``` + +每轮报告必须给出：峰值生产速度、峰值消费速度、峰值 lag、lag 清零耗时、平均/最大 batch size、DB batch p95/p99、端到端落库 p95/p99/max、重试数和死信数。 + +### 11.5 数据库落库正确性与速度 + +006 商户使用默认库 `meeting_im`，但消息表名按数据库中的数值 `tenant_id` 生成，不能直接把企业码字符串当表后缀。准备阶段先从租户注册信息解析并记录 006 的真实 `tenant_id`，再生成只读核对 SQL。 + +当前消息存储位置： + +- P2P：`im__messages`，例如真实 tenant ID 为 6 时是 `im_006_messages`。 +- 群聊：`im__group_messages_YYYYMM`。 +- 群消息分片注册：`im_group_message_shards`。 +- Kafka 消费幂等状态：Redis `kafka_consume` namespace，TTL 7 天。 +- Kafka 死信：MySQL `im_kafka_dead_letters`。 + +正式核对优先使用 WebSocket 历史接口，SQL 作为服务端审计证据。SQL 由核对程序使用白名单表名生成，不允许把外部输入直接拼入表名。 + +P2P 落库数量示例： + +```sql +SELECT + COUNT(*) AS rows_written, + COUNT(DISTINCT client_msg_id) AS unique_client_mid, + MIN(createdat) AS first_persisted_at, + MAX(createdat) AS last_persisted_at +FROM im_006_messages +WHERE client_msg_id LIKE 'lt-20260925-s1-%'; +``` + +群消息落库数量示例： + +```sql +SELECT + COUNT(*) AS rows_written, + COUNT(DISTINCT client_msg_id) AS unique_client_mid, + MIN(created_at) AS first_persisted_at, + MAX(created_at) AS last_persisted_at +FROM im_006_group_messages_202609 +WHERE client_msg_id LIKE 'lt-20260925-s2-%'; +``` + +重复检查： + +```sql +SELECT client_msg_id, COUNT(*) AS cnt, MIN(seqid), MAX(seqid) +FROM im_006_messages +WHERE client_msg_id LIKE 'lt-20260925-s1-%' +GROUP BY client_msg_id +HAVING COUNT(*) <> 1; +``` + +群表执行同样检查。P2P 有 1000 个 Topic，不能用整张表的 `MAX(seq)-MIN(seq)+1` 判断丢失；必须以压测端保存的 `client_mid -> topic -> ack_seq` manifest 与数据库逐条比对。 + +MySQL 性能至少采集： + +- `Threads_connected`、`Threads_running`、连接使用率和连接池 wait count/duration。 +- QPS、Com_insert、事务提交/回滚、行写入速度。 +- InnoDB Buffer Pool 命中率、dirty pages、checkpoint age、log waits。 +- row lock waits/time、deadlock、慢 SQL、临时磁盘表。 +- data/log fsync、读写 IOPS、吞吐、磁盘 await、util 和剩余空间。 +- P2P 批量 INSERT、群月表批量 INSERT 的 p95/p99。 +- 主从架构下的 replication lag；压测历史回查必须明确读主还是读从。 + +数据库落库速度报告包含两个值： + +```text +db_write_throughput = 测试消息成功落库数 / 首次到最后一次 DB commit 时间 +drain_time = 最后一条 ctrl 202 到 Kafka lag=0 且全部历史可见的时间 +``` + +### 11.6 Redis 观测 + +Redis 在当前消息链路中承担 seq 分配、`client_mid` 幂等 claim、群热消息缓存、Kafka 消费幂等状态和用户级 realtime fanout。需要采集： + +- `instantaneous_ops_per_sec`、命令总量与各命令耗时。 +- connected/blocked clients、连接拒绝和连接创建速度。 +- used memory、RSS、碎片率、maxmemory 使用率。 +- evicted/expired keys、keyspace hit/miss。 +- network input/output bytes、Pub/Sub channel/pattern/subscriber 数。 +- Redis 主线程 CPU、事件循环延迟、slowlog 数量。 +- Sentinel/主从环境的 role、replication offset、replication lag、failover。 +- `seq_reserve`、`unique_claim`、`hot_cache`、`kafka_consume` 操作错误和 p95/p99。 + +压测期间禁止使用无界 `KEYS` 或全库 `SCAN` 做统计，这会污染结果。Kafka 消费状态只通过应用指标抽样统计，不逐条扫描 Redis key。 + +常见异常判定： + +- evicted keys 或 rejected connections 增加：直接判失败。 +- used memory 持续增长且排空后不回落：检查 hot message、幂等 key TTL 和消费记录泄漏。 +- p99 command latency 超过 10ms：标记风险；超过 50ms 且与消息延迟尖刺相关时判失败。 +- replication lag 持续扩大或发生 failover：本轮单独标记，不与无故障基线混合。 + +### 11.7 Pod、容器与节点资源 + +每个 IM Pod 单独展示，不只看 namespace 汇总： + +```promql +sum by (pod) ( + rate(container_cpu_usage_seconds_total{namespace="im-server-system",container="im-server"}[1m]) +) +``` + +```promql +sum by (pod) ( + container_memory_working_set_bytes{namespace="im-server-system",container="im-server"} +) +``` + +```promql +sum by (pod) ( + rate(container_network_receive_bytes_total{namespace="im-server-system",pod=~"im-server-.*"}[1m]) +) +``` + +```promql +sum by (pod) ( + rate(container_network_transmit_bytes_total{namespace="im-server-system",pod=~"im-server-.*"}[1m]) +) +``` + +必须采集： + +- CPU 使用核数、CPU limit 使用率、CFS throttled seconds/periods。 +- memory working set、RSS、cache、limit 使用率、OOM 和 page fault。 +- 容器文件系统 read/write bytes、IOPS、IO wait；若根文件系统不承载业务数据，也要观察日志写入压力。 +- 网络接收/发送 bytes、packets、errors、drops、TCP retransmit。 +- Pod restart、readiness、liveness、Pending/NotReady 时间。 +- Go goroutine、heap、stack、GC 次数、GC pause、进程文件描述符。 +- `LiveSessions`、`LiveTopics`、`LiveClusterNodes`、Cluster leader 和代理 Topic 数。 +- Node CPU steal/iowait、内存压力、磁盘压力、网络错误、conntrack 使用率。 + +资源报告既要给峰值，也要给排空后是否回落。单个 Pod 明显高于其他 Pod 时，结合连接分布、Topic master 位置和网络流量判断是否发生热点。 + +建议风险线： + +- Pod CPU 持续超过 limit 的 80%，或 throttling ratio 超过 10%。 +- memory working set 超过 limit 的 80%，或排空 10 分钟后仍持续增长。 +- 节点磁盘 util 持续超过 80%，await 持续超过 20ms。 +- 网络 drop/error 或 TCP retransmit 明显增长。 +- 任一 Pod restart/OOM/NotReady，直接判本轮失败。 + +### 11.8 网络时延与压测机资源 + +两台压测机分别持续测量： + +- 到各 Pod 直连入口和负载均衡入口的 DNS、TCP connect、TLS handshake、HTTP `/healthz` RTT。 +- WebSocket Ping/Pong RTT p50/p95/p99/max，每 5 秒一次，不能用业务 `note kp` 代替协议 Ping。 +- 到 Kafka、Redis、MySQL 服务地址的 TCP connect RTT；只做低频只读探测，不绕过业务链路写数据。 +- 两台压测机之间的 RTT、丢包和 NTP offset。 +- 压测进程 CPU、RSS、GC、goroutine、FD、socket、send/receive bytes、read-loop queue depth。 + +压力生成端出现 CPU > 70%、网卡 > 70%、FD 接近上限、read-loop backlog 或 GC 尖刺时，本轮标记 `LOAD_GENERATOR_SATURATED`，不能把客户端处理不过来算成服务端丢消息。 + +### 11.9 现有 fanout 与 Cluster 查询 + +每轮开始前记录 counter 快照，结束后记录增量： + +```promql +sum by (stage, kind, target_type, status) ( + increase(realtime_fanout_events_total[10m]) +) +``` + +```promql +histogram_quantile( + 0.99, + sum by (le, kind, target_type) ( + rate(realtime_fanout_delivery_duration_seconds_bucket[5m]) + ) +) +``` + +Topic 消息使用原生 Cluster 时，`publish/skip_native_cluster` 增加属于正常现象。重点检查 Cluster 连接数、route/proxy 错误、`node for topic not found`、gob 编解码错误、Hub 队列满和 WS 写失败。 + +### 11.10 日志、快照与关联规则 + +压测日志统一带 `run_id`、scenario、machine、account_index 和脱敏 Topic hash。服务端日志不新增用户 ID 或 `client_mid` 的高基数 Prometheus label；需要定位时通过压测端保存的 manifest 关联。 + +每轮保存： + +- Prometheus 查询结果或 Grafana dashboard snapshot。 +- IM Pod、Kafka、Redis、MySQL 在测试窗口内的错误日志。 +- Kafka consumer group 压测前、峰值、结束、排空后的 offset/lag 快照。 +- Redis INFO、SLOWLOG 摘要和 exporter 指标快照。 +- MySQL GLOBAL STATUS、InnoDB status 摘要、慢 SQL 摘要和表行数核对。 +- `kubectl top pod/node`、Pod restart/condition/event 和容器日志摘要。 +- 两台压测机 node/process 指标及 agent summary。 + +日志采集必须过滤 Token、密码、API Key、Kafka/Redis/MySQL 凭证和完整消息正文。 + +### 11.11 建议仪表盘布局 + +一张压测 dashboard 分为六行： + +1. 业务结果：attempt、ACK、实时接收、历史可见、丢失、重复、乱序、p95/p99。 +2. IM Pod：连接数、发布/下行速率、Cluster route、CPU、内存、GC、goroutine。 +3. Kafka：producer/consumer rate、lag、batch、commit、retry、dead letter、broker health。 +4. Redis：OPS、command latency、memory、clients、network、eviction、replication。 +5. MySQL：insert/transaction rate、DB batch latency、pool wait、locks、Buffer Pool、IO。 +6. 基础设施：Pod/Node CPU、memory、disk IO、network、压测机资源和网络 RTT。 + +所有面板至少按 Pod、Kafka Topic/consumer group、Redis instance、MySQL instance 和压测机分组；不得只展示总和。 + +## 12. 压测程序结构 + +现有 `loadtest/realtime_fanout_probe.go` 只支持单群和单进程，不能直接用于本方案。新工具使用 Go 实现并提供六个子命令： + +```text +im-pressure prepare 注册、好友、群和 manifest +im-pressure verify 只读核验 2000 账号、1000 对关系和 2000 群成员 +im-pressure coordinator 协调双机 READY、统一起跑时间、ACK 台账与结果上报 +im-pressure agent 在每台压测机运行连接与场景 +im-pressure collect 抓取 Prometheus、K8s 和依赖组件快照 +im-pressure merge 合并两台机器结果并生成报告 +``` + +推荐文件： + +```text +loadtest/im-pressure/config.example.yaml +loadtest/im-pressure/cmd/im-pressure/ +loadtest/im-pressure/internal/protocol/ +loadtest/im-pressure/internal/prepare/ +loadtest/im-pressure/internal/runner/ +loadtest/im-pressure/internal/collector/ +loadtest/im-pressure/internal/report/ +``` + +协调方式：Machine A 启动 coordinator，等待 A/B 两个 agent 都上报 1000 个 READY，然后发布一个至少晚于当前时间 30 秒的 UTC `start_at`。两台机器按相同 `run_id` 和确定性调度执行。任何 agent 失联都将本轮标记为 `INVALID`。 + +`agent` 自身暴露 `/metrics`，提供连接状态、发布、接收、重复、乱序、断连及 Go/process 指标；ACK 和投递延迟分布写入最终 summary。`collect` 只执行只读查询，并把各数据源的原始快照与查询时间一起保存；连接凭证通过环境变量传入，不写入结果目录。 + +## 13. 结果文件 + +每轮保留以下文件： + +```text +results//config.redacted.json +results//manifest.sha256 +results//agent-a-summary.json +results//agent-b-summary.json +results//publish.jsonl +results//missing.jsonl +results//duplicate.jsonl +results//history-reconcile.json +results//db-reconcile-p2p.json +results//db-reconcile-group.json +results//timeline.json +results//metrics/im-server.prom +results//metrics/load-agent.prom +results//metrics/kafka.json +results//metrics/redis.json +results//metrics/mysql.json +results//metrics/kubernetes.json +results//metrics/network.json +results//logs/im-server-errors.log +results//logs/kafka-errors.log +results//logs/mysql-slow-summary.log +results//grafana-snapshot.json +results//report.md +``` + +缺失明细至少包含 `run_id`、场景、Topic 脱敏哈希、发送者索引、接收者索引、`client_mid`、ACK `seq`、发送时间、最后等待时间、发送 Pod 和接收 Pod。不得写入 Token、密码、API Key 或完整手机号。 + +最终 `report.md` 必须分别给出： + +- 压测业务量和消息正确性结论。 +- 实时投递与历史恢复结论。 +- Kafka 生产、消费、lag、排空和死信结论。 +- MySQL 落库数量、吞吐、耗时、锁与 IO 结论。 +- Redis OPS、耗时、内存、幂等/seq 错误和 fanout 结论。 +- 每个 Pod/Node 以及两台压测机的资源峰值和瓶颈判断。 +- 测试是否有效、是否通过，以及所有未通过项的原始证据路径。 + +## 14. 推荐执行顺序 + +1. 在测试窗口前完成环境门槛检查和 2000 账号准备。 +2. 完成 1000 对好友关系与 2000 人群，运行 `verify` 直到全部通过。 +3. 先用 20 对 P2P、100 人群做冒烟测试，验证统计口径。 +4. 执行 P0 热身。 +5. 执行 S1，等待服务端和 Kafka 恢复基线至少 5 分钟。 +6. 执行 S2，等待恢复基线至少 10 分钟。 +7. 执行 S3。 +8. 合并结果，按实时链路与持久化链路分别给出结论。 +9. 全部稳定后再单独进行 Pod 重启、断网或 Kafka/Redis 故障注入，不与本轮无故障基线混跑。 diff --git a/docs/IM-多开跨Pod实时投递压测方案.md b/docs/IM-多开跨Pod实时投递压测方案.md index e1b091d..8dedbc4 100644 --- a/docs/IM-多开跨Pod实时投递压测方案.md +++ b/docs/IM-多开跨Pod实时投递压测方案.md @@ -2,6 +2,9 @@ 日期：2026-09-13 +> 006 商户、2000 账号、两台压测机的可执行场景与统计口径见 +> [IM-006商户双机消息压力测试方案](./IM-006商户双机消息压力测试方案.md)。本文保留为通用阶梯与指标说明。 + ## 1. 目标 验证 IM Server 使用 Redis realtime fanout 后，在多 Pod、多端登录、群消息和群业务事件场景下的实时投递能力。 diff --git a/loadtest/im-pressure/.gitignore b/loadtest/im-pressure/.gitignore new file mode 100644 index 0000000..5c17dad --- /dev/null +++ b/loadtest/im-pressure/.gitignore @@ -0,0 +1,7 @@ +/im-pressure +/config.json +/manifest*.json +/results/ +/*-summary.json +/report.json +/metrics.jsonl diff --git a/loadtest/im-pressure/README.md b/loadtest/im-pressure/README.md new file mode 100644 index 0000000..7606cd8 --- /dev/null +++ b/loadtest/im-pressure/README.md @@ -0,0 +1,123 @@ +# IM 双机压力测试工具 + +该工具按商户 `006` 的固定模型运行：两台压测机各承载 1000 个 WebSocket 用户，共 2000 个用户、1000 对好友和一个 2000 人群。 + +## 统计口径 + +- 发送成功以 `{ctrl code=2xx, params.seq>0}` 为准，形成两台机器共享的 ACK 台账。 +- 实时到达只统计同时匹配本轮 `run_id` 和 `x-im-client-mid` 的 `{data}`。 +- 群消息设置 `noecho=false`，每条服务端接受的群消息期望 2000 个连接各收到一次。 +- 实时 drain 结束后通过 `{get what=data}` 按 ACK 的 Topic/Seq 回查历史，区分跨 Pod 实时漏投与数据库持久化缺失。 +- `manifest.json` 含登录 token，文件权限为 `0600`，不得提交到 Git 或放入结果包。 + +## 构建与配置 + +两台机器使用同一版本二进制、配置和 manifest： + +```bash +go build -o loadtest/im-pressure/im-pressure ./loadtest/im-pressure +cp loadtest/im-pressure/config.example.json loadtest/im-pressure/config.json +export IM_API_KEY='...' +export IM_LOAD_PASSWORD='...' +``` + +修改 `config.json`：`http_base` 指向 IM HTTP API；`endpoints.a/b` 分别填写两台机器应直连的 WS Pod/LB 地址，多个地址时用户会轮询分配。正式测试保持 `insecure_tls=false`。 + +## 一次性准备 + +只在一台机器执行。命令可重复运行，会复用已有账号、好友和群： + +```bash +./loadtest/im-pressure/im-pressure prepare \ + -config loadtest/im-pressure/config.json \ + -manifest loadtest/im-pressure/manifest.json + +./loadtest/im-pressure/im-pressure verify \ + -config loadtest/im-pressure/config.json \ + -manifest loadtest/im-pressure/manifest.json +``` + +将 `config.json` 和 `manifest.json` 安全复制到机器 B，并执行 `chmod 600 loadtest/im-pressure/manifest.json`。 + +## 双机运行 + +先在机器 A 启动协调器。`run-id` 每轮必须唯一且不超过 24 个字符： + +```bash +mkdir -p loadtest/im-pressure/results/p2p-001 +./loadtest/im-pressure/im-pressure coordinator \ + -listen :8090 -run-id p2p-001 -scenario p2p \ + -output loadtest/im-pressure/results/p2p-001/coordinator.json +``` + +然后分别启动两个 agent。`COORDINATOR_IP` 必须是机器 B 可访问的机器 A 内网地址。 + +机器 A： + +```bash +./loadtest/im-pressure/im-pressure agent \ + -config loadtest/im-pressure/config.json \ + -manifest loadtest/im-pressure/manifest.json \ + -machine a -coordinator http://COORDINATOR_IP:8090 \ + -metrics-listen :9105 \ + -output loadtest/im-pressure/results/p2p-001/agent-a-summary.json +``` + +机器 B： + +```bash +./loadtest/im-pressure/im-pressure agent \ + -config loadtest/im-pressure/config.json \ + -manifest loadtest/im-pressure/manifest.json \ + -machine b -coordinator http://COORDINATOR_IP:8090 \ + -metrics-listen :9105 \ + -output loadtest/im-pressure/results/p2p-001/agent-b-summary.json +``` + +两个 agent 都建立 1000 条连接后，协调器下发统一的 UTC `start_at`。场景可替换为： + +- `p2p`：1000 对用户两侧同时发送，默认每人 10 条、间隔 1 秒。 +- `group`：每台机器 500 个发送者，每人 2 条；全局每 100ms 一条，共 2000 条。 +- `mixed`：300 对私聊用户每人 5 条，同时每台 250 个群发送者各发 2 条。 + +## 合并报告 + +把两个 agent summary 放到同一台机器： + +```bash +./loadtest/im-pressure/im-pressure merge \ + -a loadtest/im-pressure/results/p2p-001/agent-a-summary.json \ + -b loadtest/im-pressure/results/p2p-001/agent-b-summary.json \ + -output loadtest/im-pressure/results/p2p-001/report.json +``` + +`report.json` 给出 ACK、实时投递、重复、乱序、断连和历史持久化核验结果。当前正确性门禁要求上述错误计数全部为 0。 + +每个 agent 在 `:9105/metrics` 暴露连接、发送、接收、重复、乱序、断连以及 Go/process 指标；端口冲突时可通过 `-metrics-listen` 修改，传空字符串可关闭。 + +## 指标采集 + +`collect` 可并行抓取 Prometheus、Redis exporter、Kafka exporter、mysqld exporter、kube-state-metrics/cAdvisor 和压测机 node-exporter，原始响应按时间写入 JSONL： + +```bash +./loadtest/im-pressure/im-pressure collect \ + -duration 10m -interval 5s \ + -url im=http://PROMETHEUS_HOST/api/v1/query?query=up \ + -url redis=http://REDIS_EXPORTER/metrics \ + -url kafka=http://KAFKA_EXPORTER/metrics \ + -url mysql=http://MYSQLD_EXPORTER/metrics \ + -url node-a=http://NODE_A_EXPORTER/metrics \ + -url node-b=http://NODE_B_EXPORTER/metrics \ + -output loadtest/im-pressure/results/p2p-001/metrics.jsonl +``` + +URL 含 `&` 时需用单引号包住整个 `name=url` 参数。建议起压前 2 分钟开始采集，结束后继续 5 分钟，以观察 Kafka lag、数据库落库追平时间及资源恢复。 + +## 运行前检查 + +```bash +ulimit -n 100000 +curl -fsS http://COORDINATOR_IP:8090/v1/status +``` + +两台压测机应完成时钟同步，并确认压测出口、Ingress、Pod 与 Redis/MySQL/Kafka 连接数限制可以容纳本轮负载。协调器没有鉴权，只能监听在受控压测网络中。 diff --git a/loadtest/im-pressure/agent.go b/loadtest/im-pressure/agent.go new file mode 100644 index 0000000..b3db924 --- /dev/null +++ b/loadtest/im-pressure/agent.go @@ -0,0 +1,461 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + "log" + "os" + "path/filepath" + "regexp" + "sort" + "sync" + "sync/atomic" + "time" +) + +var runIDPattern = regexp.MustCompile(`^[a-zA-Z0-9_-]{1,24}$`) + +type agentRun struct { + cfg *Config + machine string + runID string + scenario string + clients map[int]*WSClient + recorder *deliveryRecorder + ack *DurationHistogram + attempted atomic.Int64 + rejected atomic.Int64 + ready atomic.Int64 + + mu sync.Mutex + accepted []AcceptedMessage + errors []string +} + +func validScenario(scenario string) bool { + return scenario == "p2p" || scenario == "group" || scenario == "mixed" +} + +func runAgent(ctx context.Context, cfg *Config, manifest *Manifest, machine, coordinatorURL, output, metricsListen string) error { + if machine != "a" && machine != "b" { + return fmt.Errorf("machine must be a or b") + } + apiKey, err := cfg.apiKey() + if err != nil { + return err + } + password := "" + if name := cfg.PasswordEnv; name != "" { + password = os.Getenv(name) + } else { + password = os.Getenv("IM_LOAD_PASSWORD") + } + accounts := accountsForMachine(manifest, machine) + if len(accounts) != cfg.UsersPerMachine { + return fmt.Errorf("machine %s accounts=%d expected=%d", machine, len(accounts), cfg.UsersPerMachine) + } + for i := range accounts { + accounts[i].Endpoint = cfg.Endpoints[machine][(accounts[i].Index-1)%len(cfg.Endpoints[machine])] + } + coord := newCoordinatorClient(coordinatorURL) + placeholderRunID := "pending" + run := &agentRun{cfg: cfg, machine: machine, clients: make(map[int]*WSClient), ack: NewDurationHistogram()} + run.recorder = newDeliveryRecorder(placeholderRunID, machine, len(accounts)) + stopMetrics, err := startAgentMetrics(metricsListen, run) + if err != nil { + return err + } + defer stopMetrics() + if err = run.connect(ctx, accounts, manifest.GroupTopic, apiKey, password); err != nil { + run.close() + return err + } + defer run.close() + start, err := coord.ready(ctx, AgentReady{Machine: machine, ReadyConnections: len(run.clients)}) + if err != nil { + return err + } + if !runIDPattern.MatchString(start.RunID) || !validScenario(start.Scenario) { + return fmt.Errorf("invalid coordinator start: %+v", start) + } + run.runID = start.RunID + run.scenario = start.Scenario + run.recorder.setRunID(start.RunID) + if err = waitUntil(ctx, start.StartAt); err != nil { + return err + } + startedAt := time.Now().UTC() + log.Printf("agent=%s scenario=%s started", machine, start.Scenario) + if err = run.execute(ctx, start.StartAt, manifest.GroupTopic); err != nil { + return err + } + loadEndedAt := time.Now().UTC() + local := run.acceptedCopy() + if err = coord.submitLedger(ctx, machine, local); err != nil { + return err + } + ledger, err := coord.ledger(ctx) + if err != nil { + return err + } + drain, err := run.drainDuration() + if err != nil { + return err + } + if err = waitContext(ctx, drain); err != nil { + return err + } + base := AgentSummary{ + RunID: start.RunID, Scenario: start.Scenario, Machine: machine, StartedAt: startedAt, + LoadEndedAt: loadEndedAt, FinishedAt: time.Now().UTC(), ReadyConnections: len(run.clients), + PublishAttempted: run.attempted.Load(), PublishAccepted: int64(len(local)), PublishRejected: run.rejected.Load(), + AckLatency: run.ack.Snapshot(), AcceptedLocal: local, + } + run.mu.Lock() + base.Errors = append(base.Errors, run.errors...) + run.mu.Unlock() + summary := run.recorder.summarize(base, ledger) + checked, found, persistenceMissing := run.reconcileHistory(ctx, ledger) + summary.HistoryChecked = int64(checked) + summary.HistoryFound = int64(found) + summary.PersistenceLost = int64(len(persistenceMissing)) + summary.RealtimeLostStored = run.recorder.countRealtimeLostStored(ledger) + if len(persistenceMissing) > 10000 { + summary.PersistenceMissing = append([]string(nil), persistenceMissing[:10000]...) + summary.Errors = append(summary.Errors, "persistence missing details truncated at 10000") + } else { + summary.PersistenceMissing = persistenceMissing + } + summary.FinishedAt = time.Now().UTC() + if err = writeAgentSummary(output, summary); err != nil { + return err + } + if err = coord.submitSummary(ctx, summary); err != nil { + return err + } + log.Printf("agent=%s accepted=%d expected_peer=%d received_peer=%d expected_group=%d received_group=%d lost=%d duplicates=%d", + machine, summary.PublishAccepted, summary.ExpectedPeer, summary.ReceivedPeer, + summary.ExpectedGroup, summary.ReceivedGroup, summary.RealtimeLost, summary.Duplicates) + return nil +} + +type historyQuery struct { + client *WSClient + topic string + since int64 + before int64 + limit int +} + +func (r *agentRun) reconcileHistory(ctx context.Context, ledger []AcceptedMessage) (int, int, []string) { + expected := make(map[string]AcceptedMessage) + byTopic := make(map[string][]AcceptedMessage) + for _, message := range ledger { + if message.RunID != r.runID { + continue + } + if message.Kind == "p2p" && message.PeerMachine == r.machine { + expected[message.ClientMid] = message + byTopic[message.Topic] = append(byTopic[message.Topic], message) + } else if message.Kind == "group" && r.machine == "a" { + expected[message.ClientMid] = message + byTopic[message.Topic] = append(byTopic[message.Topic], message) + } + } + if len(expected) == 0 { + return 0, 0, nil + } + r.recorder.beginHistory() + queries := make([]historyQuery, 0, len(byTopic)) + for topic, messages := range byTopic { + sort.Slice(messages, func(i, j int) bool { return messages[i].Seq < messages[j].Seq }) + client := r.clients[messages[0].PeerIndex] + if messages[0].Kind == "group" { + client = r.clients[1] + } + if client == nil { + r.addError(fmt.Sprintf("history topic=%s has no verifier client", topicHash(topic))) + continue + } + for start := 0; start < len(messages); start += 200 { + end := start + 200 + if end > len(messages) { + end = len(messages) + } + queries = append(queries, historyQuery{client: client, topic: topic, + since: messages[start].Seq, before: messages[end-1].Seq + 1, limit: 256}) + } + } + for _, query := range queries { + if err := query.client.GetData(ctx, query.topic, query.since, query.before, query.limit); err != nil { + r.addError(fmt.Sprintf("history get topic=%s range=%d-%d: %v", topicHash(query.topic), query.since, query.before, err)) + } + } + _ = waitContext(ctx, 2*time.Second) + deadline := time.Now().Add(120 * time.Second) + for { + missing := missingHistory(r.recorder, expected) + if len(missing) == 0 || time.Now().After(deadline) { + break + } + if waitContext(ctx, 5*time.Second) != nil { + break + } + for _, message := range missing { + client := r.clients[message.PeerIndex] + if message.Kind == "group" { + client = r.clients[1] + } + if client != nil { + _ = client.GetData(ctx, message.Topic, message.Seq, message.Seq+1, 1) + } + } + _ = waitContext(ctx, 500*time.Millisecond) + } + missing := missingHistory(r.recorder, expected) + missingIDs := make([]string, 0, len(missing)) + for _, message := range missing { + missingIDs = append(missingIDs, message.ClientMid) + } + sort.Strings(missingIDs) + return len(expected), len(expected) - len(missing), missingIDs +} + +func missingHistory(recorder *deliveryRecorder, expected map[string]AcceptedMessage) []AcceptedMessage { + missing := make([]AcceptedMessage, 0) + for mid, message := range expected { + if !recorder.historyFound(mid) { + missing = append(missing, message) + } + } + return missing +} + +func (r *agentRun) addError(message string) { + r.mu.Lock() + defer r.mu.Unlock() + if len(r.errors) < 1000 { + r.errors = append(r.errors, message) + } +} + +func (r *agentRun) connect(ctx context.Context, accounts []AccountRecord, groupTopic, apiKey, password string) error { + ramp, err := configDuration(r.cfg.ConnectRamp) + if err != nil { + return err + } + interval := ramp / time.Duration(len(accounts)) + var wg sync.WaitGroup + errCh := make(chan error, len(accounts)) + var mu sync.Mutex + for _, account := range accounts { + account := account + wg.Add(1) + go func() { + defer wg.Done() + client, connectErr := connectWSClient(ctx, r.cfg, account, apiKey, password, r.recorder.onData, r.recorder.onDisconnect) + if connectErr != nil { + errCh <- connectErr + return + } + if connectErr = client.Subscribe(ctx, account.P2PTopic); connectErr == nil { + connectErr = client.Subscribe(ctx, groupTopic) + } + if connectErr != nil { + client.Close() + errCh <- fmt.Errorf("subscribe %s: %w", account.Username, connectErr) + return + } + mu.Lock() + r.clients[account.Index] = client + mu.Unlock() + r.ready.Add(1) + }() + if interval > 0 { + if err = waitContext(ctx, interval); err != nil { + break + } + } + } + wg.Wait() + close(errCh) + var failures []string + for connectErr := range errCh { + if len(failures) < 20 { + failures = append(failures, connectErr.Error()) + } + } + if len(r.clients) != len(accounts) { + return fmt.Errorf("ready connections=%d expected=%d failures=%v", len(r.clients), len(accounts), failures) + } + return nil +} + +func (r *agentRun) execute(ctx context.Context, start time.Time, groupTopic string) error { + if groupTopic == "" { + return fmt.Errorf("group topic is empty") + } + switch r.scenario { + 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, 500, 2) + case "mixed": + var wg sync.WaitGroup + errCh := make(chan error, 2) + wg.Add(2) + go func() { defer wg.Done(); errCh <- r.scheduleP2P(ctx, start, 300, 5, 20*time.Second) }() + go func() { defer wg.Done(); errCh <- r.scheduleGroup(ctx, start, groupTopic, 301, 550, 2) }() + wg.Wait() + close(errCh) + for err := range errCh { + if err != nil { + return err + } + } + return nil + default: + return fmt.Errorf("unknown scenario %s", r.scenario) + } +} + +func (r *agentRun) scheduleP2P(ctx context.Context, start time.Time, users, messages int, interval time.Duration) error { + var wg sync.WaitGroup + for index := 1; index <= users; index++ { + client := r.clients[index] + if client == nil { + return fmt.Errorf("missing client %d", index) + } + wg.Add(1) + go func(index int, client *WSClient) { + defer wg.Done() + jitter := time.Duration((index*37)%200) * time.Millisecond + for messageIndex := 1; messageIndex <= messages; messageIndex++ { + target := start.Add(time.Duration(messageIndex-1)*interval + jitter) + if waitUntil(ctx, target) != nil { + return + } + mid := fmt.Sprintf("%s-p-%s-%04d-%02d", r.runID, r.machine, index, messageIndex) + accepted, err := r.publish(ctx, client, client.account.P2PTopic, mid, messageIndex) + if err == nil { + accepted.PeerIndex = index + if r.machine == "a" { + accepted.PeerMachine = "b" + } else { + accepted.PeerMachine = "a" + } + r.addAccepted(accepted) + } + } + }(index, client) + } + wg.Wait() + return ctx.Err() +} + +func (r *agentRun) scheduleGroup(ctx context.Context, start time.Time, topic string, first, last, rounds int) error { + if first < 1 || last > len(r.clients) || first > last { + return fmt.Errorf("invalid group publisher range %d-%d", first, last) + } + count := last - first + 1 + roundDuration := time.Duration(count) * 200 * time.Millisecond + offset := time.Duration(0) + if r.machine == "b" { + offset = 100 * time.Millisecond + } + var wg sync.WaitGroup + for index := first; index <= last; index++ { + client := r.clients[index] + position := index - first + wg.Add(1) + go func(index, position int, client *WSClient) { + defer wg.Done() + for round := 1; round <= rounds; round++ { + target := start.Add(time.Duration(round-1)*roundDuration + time.Duration(position)*200*time.Millisecond + offset) + if waitUntil(ctx, target) != nil { + return + } + mid := fmt.Sprintf("%s-g-%s-%04d-%02d", r.runID, r.machine, index, round) + accepted, err := r.publish(ctx, client, topic, mid, round) + if err == nil { + r.addAccepted(accepted) + } + } + }(index, position, client) + } + wg.Wait() + return ctx.Err() +} + +func (r *agentRun) publish(ctx context.Context, client *WSClient, topic, mid string, messageIndex int) (AcceptedMessage, error) { + r.attempted.Add(1) + accepted, err := client.Publish(ctx, topic, mid, r.runID, r.scenario, messageIndex) + if err != nil { + r.rejected.Add(1) + r.mu.Lock() + if len(r.errors) < 1000 { + r.errors = append(r.errors, fmt.Sprintf("publish %s: %v", mid, err)) + } + r.mu.Unlock() + return accepted, err + } + r.ack.Observe(accepted.AckAt.Sub(accepted.SentAt)) + return accepted, nil +} + +func (r *agentRun) addAccepted(message AcceptedMessage) { + r.mu.Lock() + r.accepted = append(r.accepted, message) + r.mu.Unlock() +} + +func (r *agentRun) acceptedCopy() []AcceptedMessage { + r.mu.Lock() + defer r.mu.Unlock() + out := append([]AcceptedMessage(nil), r.accepted...) + sort.Slice(out, func(i, j int) bool { return out[i].ClientMid < out[j].ClientMid }) + return out +} + +func (r *agentRun) drainDuration() (time.Duration, error) { + if r.scenario == "p2p" { + return configDuration(r.cfg.DrainP2P) + } + return configDuration(r.cfg.DrainGroup) +} + +func (r *agentRun) close() { + for _, client := range r.clients { + client.Close() + } +} + +func waitUntil(ctx context.Context, target time.Time) error { + delay := time.Until(target) + if delay <= 0 { + return nil + } + return waitContext(ctx, delay) +} + +func mustDuration(value string) time.Duration { + d, err := time.ParseDuration(value) + if err != nil || d <= 0 { + return time.Second + } + return d +} + +func writeAgentSummary(path string, summary AgentSummary) error { + raw, err := json.MarshalIndent(summary, "", " ") + if err != nil { + return err + } + if dir := filepath.Dir(path); dir != "." { + if err = os.MkdirAll(dir, 0o750); err != nil { + return err + } + } + return os.WriteFile(path, append(raw, '\n'), 0o600) +} diff --git a/loadtest/im-pressure/agent_metrics.go b/loadtest/im-pressure/agent_metrics.go new file mode 100644 index 0000000..875d337 --- /dev/null +++ b/loadtest/im-pressure/agent_metrics.go @@ -0,0 +1,68 @@ +package main + +import ( + "context" + "fmt" + "net" + "net/http" + "time" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/collectors" + "github.com/prometheus/client_golang/prometheus/promhttp" +) + +func startAgentMetrics(address string, run *agentRun) (func(), error) { + if address == "" { + return func() {}, nil + } + listener, err := net.Listen("tcp", address) + if err != nil { + return nil, fmt.Errorf("listen for agent metrics on %s: %w", address, err) + } + labels := prometheus.Labels{"machine": run.machine} + registry := prometheus.NewRegistry() + registry.MustRegister( + collectors.NewGoCollector(), + collectors.NewProcessCollector(collectors.ProcessCollectorOpts{}), + prometheus.NewGaugeFunc(prometheus.GaugeOpts{Name: "im_load_agent_ready_connections", Help: "Current ready WebSocket connections.", ConstLabels: labels}, func() float64 { + return float64(run.ready.Load()) + }), + prometheus.NewCounterFunc(prometheus.CounterOpts{Name: "im_load_agent_publish_attempted_total", Help: "Publish commands attempted.", ConstLabels: labels}, func() float64 { + return float64(run.attempted.Load()) + }), + prometheus.NewCounterFunc(prometheus.CounterOpts{Name: "im_load_agent_publish_accepted_total", Help: "Publish commands acknowledged by the server.", ConstLabels: labels}, func() float64 { + run.mu.Lock() + defer run.mu.Unlock() + return float64(len(run.accepted)) + }), + prometheus.NewCounterFunc(prometheus.CounterOpts{Name: "im_load_agent_publish_rejected_total", Help: "Publish commands rejected or timed out.", ConstLabels: labels}, func() float64 { + return float64(run.rejected.Load()) + }), + prometheus.NewCounterFunc(prometheus.CounterOpts{Name: "im_load_agent_received_total", Help: "Unique matching data deliveries received.", ConstLabels: labels}, func() float64 { + received, _, _, _ := run.recorder.metricCounts() + return float64(received) + }), + prometheus.NewCounterFunc(prometheus.CounterOpts{Name: "im_load_agent_duplicates_total", Help: "Duplicate matching deliveries received.", ConstLabels: labels}, func() float64 { + _, duplicates, _, _ := run.recorder.metricCounts() + return float64(duplicates) + }), + prometheus.NewCounterFunc(prometheus.CounterOpts{Name: "im_load_agent_out_of_order_total", Help: "Per-connection topic sequence regressions.", ConstLabels: labels}, func() float64 { + _, _, outOfOrder, _ := run.recorder.metricCounts() + return float64(outOfOrder) + }), + prometheus.NewCounterFunc(prometheus.CounterOpts{Name: "im_load_agent_disconnects_total", Help: "Unexpected WebSocket disconnects.", ConstLabels: labels}, func() float64 { + _, _, _, disconnects := run.recorder.metricCounts() + return float64(disconnects) + }), + ) + mux := http.NewServeMux() + mux.Handle("/metrics", promhttp.HandlerFor(registry, promhttp.HandlerOpts{})) + server := &http.Server{Handler: mux, ReadHeaderTimeout: 5 * time.Second} + go func() { _ = server.Serve(listener) }() + return func() { + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + _ = server.Shutdown(ctx) + }, nil +} diff --git a/loadtest/im-pressure/collector.go b/loadtest/im-pressure/collector.go new file mode 100644 index 0000000..78d8fb7 --- /dev/null +++ b/loadtest/im-pressure/collector.go @@ -0,0 +1,114 @@ +package main + +import ( + "bufio" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "os" + "path/filepath" + "strings" + "sync" + "time" +) + +type namedURLs []string + +func (n *namedURLs) String() string { return strings.Join(*n, ",") } +func (n *namedURLs) Set(value string) error { + if !strings.Contains(value, "=") { + return fmt.Errorf("URL must use name=https://host/path") + } + *n = append(*n, value) + return nil +} + +type metricSnapshot struct { + Timestamp time.Time `json:"timestamp"` + Name string `json:"name"` + URL string `json:"url"` + Status int `json:"status"` + Duration int64 `json:"duration_ms"` + Body string `json:"body,omitempty"` + Error string `json:"error,omitempty"` +} + +func runCollector(ctx context.Context, sources namedURLs, output string, interval, duration time.Duration) error { + if len(sources) == 0 { + return fmt.Errorf("at least one --url name=url is required") + } + if interval <= 0 || duration <= 0 { + return fmt.Errorf("interval and duration must be positive") + } + if dir := filepath.Dir(output); dir != "." { + if err := os.MkdirAll(dir, 0o750); err != nil { + return err + } + } + file, err := os.OpenFile(output, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0o600) + if err != nil { + return err + } + defer file.Close() + writer := bufio.NewWriterSize(file, 256<<10) + defer writer.Flush() + client := &http.Client{Timeout: 10 * time.Second} + deadline := time.Now().Add(duration) + for { + snapshots := collectOnce(ctx, client, sources) + for _, snapshot := range snapshots { + raw, _ := json.Marshal(snapshot) + if _, err = writer.Write(append(raw, '\n')); err != nil { + return err + } + } + if err = writer.Flush(); err != nil { + return err + } + if time.Now().Add(interval).After(deadline) { + return nil + } + if err = waitContext(ctx, interval); err != nil { + return err + } + } +} + +func collectOnce(ctx context.Context, client *http.Client, sources namedURLs) []metricSnapshot { + out := make([]metricSnapshot, len(sources)) + var wg sync.WaitGroup + for i, source := range sources { + i, source := i, source + wg.Add(1) + go func() { + defer wg.Done() + parts := strings.SplitN(source, "=", 2) + snapshot := metricSnapshot{Timestamp: time.Now().UTC(), Name: parts[0], URL: parts[1]} + started := time.Now() + req, err := http.NewRequestWithContext(ctx, http.MethodGet, parts[1], nil) + if err == nil { + var response *http.Response + response, err = client.Do(req) + if err == nil { + snapshot.Status = response.StatusCode + raw, readErr := io.ReadAll(io.LimitReader(response.Body, 16<<20)) + _ = response.Body.Close() + if readErr != nil { + err = readErr + } else { + snapshot.Body = string(raw) + } + } + } + snapshot.Duration = time.Since(started).Milliseconds() + if err != nil { + snapshot.Error = err.Error() + } + out[i] = snapshot + }() + } + wg.Wait() + return out +} diff --git a/loadtest/im-pressure/config.example.json b/loadtest/im-pressure/config.example.json new file mode 100644 index 0000000..77fd4c5 --- /dev/null +++ b/loadtest/im-pressure/config.example.json @@ -0,0 +1,19 @@ +{ + "tenant": "006", + "http_base": "https://im-api.example.com", + "api_key_env": "IM_API_KEY", + "password_env": "IM_LOAD_PASSWORD", + "users_per_machine": 1000, + "endpoints": { + "a": ["wss://im-ws-pod-a.example.com/v0/channels"], + "b": ["wss://im-ws-pod-b.example.com/v0/channels"] + }, + "insecure_tls": false, + "prepare_workers": 20, + "connect_ramp": "20s", + "command_timeout": "20s", + "p2p_messages_per_user": 10, + "p2p_interval": "1s", + "drain_p2p": "30s", + "drain_group": "60s" +} diff --git a/loadtest/im-pressure/coordinator.go b/loadtest/im-pressure/coordinator.go new file mode 100644 index 0000000..4cb0c65 --- /dev/null +++ b/loadtest/im-pressure/coordinator.go @@ -0,0 +1,323 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + "io" + "log" + "net/http" + "os" + "path/filepath" + "sort" + "strings" + "sync" + "time" +) + +type StartInfo struct { + RunID string `json:"run_id"` + Scenario string `json:"scenario"` + StartAt time.Time `json:"start_at"` +} + +type AgentReady struct { + Machine string `json:"machine"` + ReadyConnections int `json:"ready_connections"` +} + +type coordinatorState struct { + mu sync.Mutex + start StartInfo + startDelay time.Duration + expectedAgents int + ready map[string]AgentReady + ledgers map[string][]AcceptedMessage + summaries map[string]AgentSummary + complete chan struct{} + completeOnce sync.Once +} + +func runCoordinator(ctx context.Context, listen, runID, scenario, output string, expectedAgents int, startDelay time.Duration) error { + if !runIDPattern.MatchString(runID) { + return fmt.Errorf("run-id must match %s", runIDPattern.String()) + } + if !validScenario(scenario) { + return fmt.Errorf("unsupported scenario %q", scenario) + } + if expectedAgents <= 0 { + expectedAgents = 2 + } + state := &coordinatorState{ + start: StartInfo{RunID: runID, Scenario: scenario}, startDelay: startDelay, expectedAgents: expectedAgents, + ready: make(map[string]AgentReady), ledgers: make(map[string][]AcceptedMessage), + summaries: make(map[string]AgentSummary), complete: make(chan struct{}), + } + mux := http.NewServeMux() + mux.HandleFunc("/v1/ready", state.handleReady) + mux.HandleFunc("/v1/start", state.handleStart) + mux.HandleFunc("/v1/ledger", state.handleLedger) + mux.HandleFunc("/v1/summary", state.handleSummary) + mux.HandleFunc("/v1/status", state.handleStatus) + server := &http.Server{Addr: listen, Handler: mux, ReadHeaderTimeout: 5 * time.Second} + errCh := make(chan error, 1) + go func() { + log.Printf("coordinator listening=%s run_id=%s scenario=%s expected_agents=%d", listen, runID, scenario, expectedAgents) + if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed { + errCh <- err + } + }() + select { + case <-ctx.Done(): + case err := <-errCh: + return err + case <-state.complete: + log.Printf("all agent summaries received") + } + shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _ = server.Shutdown(shutdownCtx) + if output != "" { + if err := state.writeOutput(output); err != nil { + return err + } + } + return nil +} + +func (s *coordinatorState) handleReady(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + var ready AgentReady + if !decodeLimitedJSON(w, r, &ready) || (ready.Machine != "a" && ready.Machine != "b") || ready.ReadyConnections <= 0 { + if ready.Machine != "a" && ready.Machine != "b" { + http.Error(w, "invalid machine", http.StatusBadRequest) + } + return + } + s.mu.Lock() + s.ready[ready.Machine] = ready + if len(s.ready) == s.expectedAgents && s.start.StartAt.IsZero() { + s.start.StartAt = time.Now().UTC().Add(s.startDelay) + log.Printf("all agents ready; start_at=%s", s.start.StartAt.Format(time.RFC3339Nano)) + } + start := s.start + s.mu.Unlock() + writeJSON(w, http.StatusOK, start) +} + +func (s *coordinatorState) handleStart(w http.ResponseWriter, r *http.Request) { + s.mu.Lock() + defer s.mu.Unlock() + if s.start.StartAt.IsZero() { + writeJSON(w, http.StatusAccepted, map[string]any{"ready_agents": len(s.ready), "expected_agents": s.expectedAgents}) + return + } + writeJSON(w, http.StatusOK, s.start) +} + +func (s *coordinatorState) handleLedger(w http.ResponseWriter, r *http.Request) { + machine := strings.ToLower(strings.TrimSpace(r.URL.Query().Get("machine"))) + switch r.Method { + case http.MethodPost: + if machine != "a" && machine != "b" { + http.Error(w, "invalid machine", http.StatusBadRequest) + return + } + var messages []AcceptedMessage + if !decodeLimitedJSON(w, r, &messages) { + return + } + s.mu.Lock() + s.ledgers[machine] = messages + count := len(s.ledgers) + s.mu.Unlock() + writeJSON(w, http.StatusOK, map[string]int{"submitted_agents": count}) + case http.MethodGet: + s.mu.Lock() + defer s.mu.Unlock() + if len(s.ledgers) < s.expectedAgents { + writeJSON(w, http.StatusAccepted, map[string]int{"submitted_agents": len(s.ledgers), "expected_agents": s.expectedAgents}) + return + } + var merged []AcceptedMessage + for _, messages := range s.ledgers { + merged = append(merged, messages...) + } + sort.Slice(merged, func(i, j int) bool { return merged[i].ClientMid < merged[j].ClientMid }) + writeJSON(w, http.StatusOK, merged) + default: + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + } +} + +func (s *coordinatorState) handleSummary(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + var summary AgentSummary + if !decodeLimitedJSON(w, r, &summary) { + return + } + if summary.Machine != "a" && summary.Machine != "b" { + http.Error(w, "invalid machine", http.StatusBadRequest) + return + } + s.mu.Lock() + s.summaries[summary.Machine] = summary + complete := len(s.summaries) == s.expectedAgents + s.mu.Unlock() + writeJSON(w, http.StatusOK, map[string]any{"complete": complete}) + if complete { + s.completeOnce.Do(func() { + go func() { + time.Sleep(500 * time.Millisecond) + close(s.complete) + }() + }) + } +} + +func (s *coordinatorState) handleStatus(w http.ResponseWriter, _ *http.Request) { + s.mu.Lock() + defer s.mu.Unlock() + writeJSON(w, http.StatusOK, map[string]any{ + "start": s.start, "ready": s.ready, "ledger_agents": len(s.ledgers), "summary_agents": len(s.summaries), + }) +} + +func (s *coordinatorState) writeOutput(path string) error { + s.mu.Lock() + data := struct { + Start StartInfo `json:"start"` + Ready map[string]AgentReady `json:"ready"` + Ledgers map[string][]AcceptedMessage `json:"ledgers"` + Summaries map[string]AgentSummary `json:"summaries"` + }{s.start, s.ready, s.ledgers, s.summaries} + s.mu.Unlock() + raw, err := json.MarshalIndent(data, "", " ") + if err != nil { + return err + } + if err = os.MkdirAll(filepath.Dir(path), 0o750); err != nil && filepath.Dir(path) != "." { + return err + } + return os.WriteFile(path, append(raw, '\n'), 0o600) +} + +func decodeLimitedJSON(w http.ResponseWriter, r *http.Request, output any) bool { + dec := json.NewDecoder(io.LimitReader(r.Body, 32<<20)) + if err := dec.Decode(output); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return false + } + return true +} + +func writeJSON(w http.ResponseWriter, status int, value any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(value) +} + +type coordinatorClient struct { + base string + client *http.Client +} + +func newCoordinatorClient(base string) *coordinatorClient { + return &coordinatorClient{base: strings.TrimRight(base, "/"), client: &http.Client{Timeout: 35 * time.Second}} +} + +func (c *coordinatorClient) ready(ctx context.Context, ready AgentReady) (StartInfo, error) { + var start StartInfo + if _, err := c.request(ctx, http.MethodPost, "/v1/ready", ready, &start); err != nil { + return start, err + } + for start.StartAt.IsZero() { + if err := waitContext(ctx, time.Second); err != nil { + return start, err + } + status, err := c.request(ctx, http.MethodGet, "/v1/start", nil, &start) + if err != nil { + return start, err + } + if status != http.StatusOK { + start = StartInfo{} + } + } + return start, nil +} + +func (c *coordinatorClient) submitLedger(ctx context.Context, machine string, messages []AcceptedMessage) error { + _, err := c.request(ctx, http.MethodPost, "/v1/ledger?machine="+machine, messages, nil) + return err +} + +func (c *coordinatorClient) ledger(ctx context.Context) ([]AcceptedMessage, error) { + for { + var messages []AcceptedMessage + status, err := c.request(ctx, http.MethodGet, "/v1/ledger", nil, &messages) + if err != nil { + return nil, err + } + if status == http.StatusOK { + return messages, nil + } + if err = waitContext(ctx, time.Second); err != nil { + return nil, err + } + } +} + +func (c *coordinatorClient) submitSummary(ctx context.Context, summary AgentSummary) error { + _, err := c.request(ctx, http.MethodPost, "/v1/summary", summary, nil) + return err +} + +func (c *coordinatorClient) request(ctx context.Context, method, path string, input, output any) (int, error) { + var body io.Reader + if input != nil { + raw, err := json.Marshal(input) + if err != nil { + return 0, err + } + body = strings.NewReader(string(raw)) + } + req, err := http.NewRequestWithContext(ctx, method, c.base+path, body) + if err != nil { + return 0, err + } + if input != nil { + req.Header.Set("Content-Type", "application/json") + } + resp, err := c.client.Do(req) + if err != nil { + return 0, err + } + defer resp.Body.Close() + if resp.StatusCode >= 400 { + raw, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) + return resp.StatusCode, fmt.Errorf("coordinator HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(raw))) + } + if output != nil && resp.StatusCode == http.StatusOK { + if err = json.NewDecoder(resp.Body).Decode(output); err != nil { + return resp.StatusCode, err + } + } + return resp.StatusCode, nil +} + +func waitContext(ctx context.Context, delay time.Duration) error { + timer := time.NewTimer(delay) + defer timer.Stop() + select { + case <-ctx.Done(): + return ctx.Err() + case <-timer.C: + return nil + } +} diff --git a/loadtest/im-pressure/coordinator_test.go b/loadtest/im-pressure/coordinator_test.go new file mode 100644 index 0000000..0db3bed --- /dev/null +++ b/loadtest/im-pressure/coordinator_test.go @@ -0,0 +1,60 @@ +package main + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + "time" +) + +func TestCoordinatorClientWaitsForBothAgentsAndMergesLedger(t *testing.T) { + state := &coordinatorState{ + start: StartInfo{RunID: "run1", Scenario: "p2p"}, startDelay: time.Second, expectedAgents: 2, + ready: make(map[string]AgentReady), ledgers: make(map[string][]AcceptedMessage), + summaries: make(map[string]AgentSummary), complete: make(chan struct{}), + } + server := httptest.NewServer(testCoordinatorMux(state)) + defer server.Close() + client := newCoordinatorClient(server.URL) + + aDone := make(chan StartInfo, 1) + go func() { + start, _ := client.ready(context.Background(), AgentReady{Machine: "a", ReadyConnections: 1000}) + aDone <- start + }() + time.Sleep(50 * time.Millisecond) + startB, err := client.ready(context.Background(), AgentReady{Machine: "b", ReadyConnections: 1000}) + if err != nil || startB.StartAt.IsZero() { + t.Fatalf("machine B ready: start=%+v err=%v", startB, err) + } + select { + case startA := <-aDone: + if !startA.StartAt.Equal(startB.StartAt) { + t.Fatalf("different start times A=%s B=%s", startA.StartAt, startB.StartAt) + } + case <-time.After(2 * time.Second): + t.Fatal("machine A did not receive start barrier") + } + + if err = client.submitLedger(context.Background(), "a", []AcceptedMessage{{ClientMid: "b"}}); err != nil { + t.Fatal(err) + } + if err = client.submitLedger(context.Background(), "b", []AcceptedMessage{{ClientMid: "a"}}); err != nil { + t.Fatal(err) + } + ledger, err := client.ledger(context.Background()) + if err != nil || len(ledger) != 2 || ledger[0].ClientMid != "a" || ledger[1].ClientMid != "b" { + t.Fatalf("ledger=%+v err=%v", ledger, err) + } +} + +func testCoordinatorMux(state *coordinatorState) *http.ServeMux { + mux := http.NewServeMux() + mux.HandleFunc("/v1/ready", state.handleReady) + mux.HandleFunc("/v1/start", state.handleStart) + mux.HandleFunc("/v1/ledger", state.handleLedger) + mux.HandleFunc("/v1/summary", state.handleSummary) + mux.HandleFunc("/v1/status", state.handleStatus) + return mux +} diff --git a/loadtest/im-pressure/histogram.go b/loadtest/im-pressure/histogram.go new file mode 100644 index 0000000..5e7958b --- /dev/null +++ b/loadtest/im-pressure/histogram.go @@ -0,0 +1,64 @@ +package main + +import ( + "sort" + "sync" + "time" +) + +var latencyBounds = []time.Duration{ + time.Millisecond, 3 * time.Millisecond, 5 * time.Millisecond, 10 * time.Millisecond, + 25 * time.Millisecond, 50 * time.Millisecond, 100 * time.Millisecond, 200 * time.Millisecond, + 500 * time.Millisecond, time.Second, 2 * time.Second, 5 * time.Second, 10 * time.Second, +} + +type DurationHistogram struct { + mu sync.Mutex + count int64 + sum time.Duration + max time.Duration + bucket []int64 +} + +type HistogramSnapshot struct { + Count int64 `json:"count"` + SumMS float64 `json:"sum_ms"` + MaxMS float64 `json:"max_ms"` + Buckets map[string]int64 `json:"buckets"` +} + +func NewDurationHistogram() *DurationHistogram { + return &DurationHistogram{bucket: make([]int64, len(latencyBounds)+1)} +} + +func (h *DurationHistogram) Observe(value time.Duration) { + if value < 0 { + return + } + h.mu.Lock() + defer h.mu.Unlock() + h.count++ + h.sum += value + if value > h.max { + h.max = value + } + idx := sort.Search(len(latencyBounds), func(i int) bool { return value <= latencyBounds[i] }) + h.bucket[idx]++ +} + +func (h *DurationHistogram) Snapshot() HistogramSnapshot { + h.mu.Lock() + defer h.mu.Unlock() + buckets := make(map[string]int64, len(h.bucket)) + for i, count := range h.bucket { + label := "+Inf" + if i < len(latencyBounds) { + label = latencyBounds[i].String() + } + buckets[label] = count + } + return HistogramSnapshot{ + Count: h.count, SumMS: float64(h.sum) / float64(time.Millisecond), + MaxMS: float64(h.max) / float64(time.Millisecond), Buckets: buckets, + } +} diff --git a/loadtest/im-pressure/httpapi.go b/loadtest/im-pressure/httpapi.go new file mode 100644 index 0000000..21072bf --- /dev/null +++ b/loadtest/im-pressure/httpapi.go @@ -0,0 +1,259 @@ +package main + +import ( + "bytes" + "context" + "crypto/tls" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "strconv" + "strings" + "sync/atomic" + "time" +) + +type apiClient struct { + baseURL string + apiKey string + client *http.Client + seq atomic.Uint64 +} + +type apiEnvelope struct { + Code int `json:"code"` + Data json.RawMessage `json:"data"` + Reason string `json:"reason"` + Message string `json:"message"` + RequestID string `json:"request_id"` +} + +type apiError struct { + Status int + Code int + Reason string + Message string +} + +func (e *apiError) Error() string { + return fmt.Sprintf("HTTP %d code=%d reason=%s message=%s", e.Status, e.Code, e.Reason, e.Message) +} + +func newAPIClient(cfg *Config, apiKey string) *apiClient { + transport := http.DefaultTransport.(*http.Transport).Clone() + if cfg.InsecureTLS { + transport.TLSClientConfig = &tls.Config{InsecureSkipVerify: true} // #nosec G402: explicit load-test option. + } + return &apiClient{ + baseURL: strings.TrimRight(cfg.HTTPBase, "/"), apiKey: apiKey, + client: &http.Client{Timeout: 30 * time.Second, Transport: transport}, + } +} + +func (a *apiClient) do(ctx context.Context, method, path, token, idempotencyKey string, body any, output any) error { + var reader io.Reader + if body != nil { + raw, err := json.Marshal(body) + if err != nil { + return err + } + reader = bytes.NewReader(raw) + } + req, err := http.NewRequestWithContext(ctx, method, a.baseURL+path, reader) + if err != nil { + return err + } + req.Header.Set("X-MeetingIM-APIKey", a.apiKey) + req.Header.Set("X-IM-Schema", "1") + req.Header.Set("X-Request-ID", fmt.Sprintf("loadtest-%d-%d", time.Now().UnixNano(), a.seq.Add(1))) + if token != "" { + req.Header.Set("Authorization", "Bearer "+token) + } + if idempotencyKey != "" { + req.Header.Set("Idempotency-Key", idempotencyKey) + } + if body != nil { + req.Header.Set("Content-Type", "application/json") + } + resp, err := a.client.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + raw, err := io.ReadAll(io.LimitReader(resp.Body, 8<<20)) + if err != nil { + return err + } + var envelope apiEnvelope + if err = json.Unmarshal(raw, &envelope); err != nil { + return fmt.Errorf("decode HTTP %d response: %w", resp.StatusCode, err) + } + if resp.StatusCode < 200 || resp.StatusCode >= 300 || envelope.Code != 0 { + return &apiError{Status: resp.StatusCode, Code: envelope.Code, Reason: envelope.Reason, Message: envelope.Message} + } + if output != nil && len(envelope.Data) > 0 { + if err = json.Unmarshal(envelope.Data, output); err != nil { + return fmt.Errorf("decode response data: %w", err) + } + } + return nil +} + +type privacyResponse struct { + FriendRequestVerification bool `json:"friend_request_verification"` + GroupInviteVerification bool `json:"group_invite_verification"` + Version int `json:"version"` +} + +func (a *apiClient) disableGroupInviteVerification(ctx context.Context, account AccountRecord) error { + var privacy privacyResponse + if err := a.do(ctx, http.MethodGet, "/v1/im/me/privacy", account.Token, "", nil, &privacy); err != nil { + return err + } + if !privacy.GroupInviteVerification { + return nil + } + body := map[string]any{"group_invite_verification": false, "version": privacy.Version} + return a.do(ctx, http.MethodPatch, "/v1/im/me/privacy", account.Token, + "loadtest-privacy-"+account.Username, body, &privacy) +} + +type friendRequestResponse struct { + RequestID string `json:"request_id"` + Status string `json:"status"` + Topic string `json:"topic"` +} + +type friendRequestListResponse struct { + Items []struct { + RequestID string `json:"request_id"` + ApplicantID string `json:"applicant_id"` + Status string `json:"status"` + } `json:"items"` + NextCursor string `json:"next_cursor"` +} + +func (a *apiClient) findPendingFriendRequest(ctx context.Context, target AccountRecord, applicantID string) (string, error) { + cursor := "" + for { + path := "/v1/im/friend-requests?direction=inbound&status=pending&limit=100" + if cursor != "" { + path += "&cursor=" + url.QueryEscape(cursor) + } + var page friendRequestListResponse + if err := a.do(ctx, http.MethodGet, path, target.Token, "", nil, &page); err != nil { + return "", err + } + for _, item := range page.Items { + if item.ApplicantID == applicantID && item.Status == "pending" { + return item.RequestID, nil + } + } + if page.NextCursor == "" || page.NextCursor == cursor { + return "", nil + } + cursor = page.NextCursor + } +} + +func (a *apiClient) createFriendRequest(ctx context.Context, source, target AccountRecord) (*friendRequestResponse, error) { + var result friendRequestResponse + body := map[string]any{"target_user_id": target.UserID, "message": "loadtest", "source": "search", "source_topic": ""} + err := a.do(ctx, http.MethodPost, "/v1/im/friend-requests", source.Token, + fmt.Sprintf("loadtest-friend-%04d", source.Index), body, &result) + return &result, err +} + +func (a *apiClient) acceptFriendRequest(ctx context.Context, target AccountRecord, requestID string) (*friendRequestResponse, error) { + var result friendRequestResponse + err := a.do(ctx, http.MethodPost, "/v1/im/friend-requests/"+url.PathEscape(requestID)+"/accept", target.Token, + "loadtest-friend-accept-"+requestID, map[string]any{}, &result) + return &result, err +} + +type p2pResponse struct { + Topic string `json:"topic"` + Relationship string `json:"relationship"` +} + +func (a *apiClient) ensureP2P(ctx context.Context, source, target AccountRecord) (*p2pResponse, error) { + var result p2pResponse + err := a.do(ctx, http.MethodPost, "/v1/im/conversations/p2p", source.Token, + fmt.Sprintf("loadtest-p2p-%s-%04d", source.Machine, source.Index), + map[string]any{"peer_user_id": target.UserID}, &result) + return &result, err +} + +type groupCreateResponse struct { + Topic string `json:"topic"` + MemberCount int `json:"member_count"` +} + +func (a *apiClient) createGroup(ctx context.Context, owner AccountRecord) (*groupCreateResponse, error) { + var result groupCreateResponse + body := map[string]any{"name": "IM 006 Load Test", "description": "im-pressure", "member_user_ids": []string{}} + err := a.do(ctx, http.MethodPost, "/v1/im/groups", owner.Token, "loadtest-group-create", body, &result) + return &result, err +} + +type groupInvitationItem struct { + UserID string `json:"user_id"` + Status string `json:"status"` + Reason string `json:"reason"` +} + +type groupInvitationResponse struct { + Topic string `json:"topic"` + Items []groupInvitationItem `json:"items"` +} + +func (a *apiClient) inviteGroup(ctx context.Context, owner AccountRecord, topic string, batch int, userIDs []string) (*groupInvitationResponse, error) { + var result groupInvitationResponse + err := a.do(ctx, http.MethodPost, "/v1/im/groups/"+url.PathEscape(topic)+"/invitations", owner.Token, + fmt.Sprintf("loadtest-group-%s-invite-%03d", topicHash(topic), batch), map[string]any{"user_ids": userIDs, "message": "loadtest"}, &result) + return &result, err +} + +type groupMembersResponse struct { + Items []struct { + UserID string `json:"user_id"` + } `json:"items"` + NextCursor string `json:"next_cursor"` + HasMore bool `json:"has_more"` +} + +func (a *apiClient) countGroupMembers(ctx context.Context, account AccountRecord, topic string) (int, error) { + cursor := "" + seen := make(map[string]bool) + for { + path := "/v1/im/groups/" + url.PathEscape(topic) + "/members?limit=100" + if cursor != "" { + path += "&cursor=" + url.QueryEscape(cursor) + } + var page groupMembersResponse + if err := a.do(ctx, http.MethodGet, path, account.Token, "", nil, &page); err != nil { + return 0, err + } + for _, item := range page.Items { + seen[item.UserID] = true + } + if page.NextCursor == "" || (!page.HasMore && len(page.Items) < 100) { + break + } + if page.NextCursor == cursor { + return 0, fmt.Errorf("group member cursor did not advance: %s", cursor) + } + cursor = page.NextCursor + } + return len(seen), nil +} + +func queryInt(value string, fallback int) int { + n, err := strconv.Atoi(value) + if err != nil { + return fallback + } + return n +} diff --git a/loadtest/im-pressure/main.go b/loadtest/im-pressure/main.go new file mode 100644 index 0000000..87dbfe5 --- /dev/null +++ b/loadtest/im-pressure/main.go @@ -0,0 +1,146 @@ +package main + +import ( + "context" + "flag" + "fmt" + "log" + "os" + "os/signal" + "syscall" + "time" +) + +func main() { + log.SetFlags(log.LstdFlags | log.Lmicroseconds | log.LUTC) + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + if len(os.Args) < 2 { + usage() + os.Exit(2) + } + var err error + switch os.Args[1] { + case "prepare": + err = prepareCommand(ctx, os.Args[2:]) + case "verify": + err = verifyCommand(ctx, os.Args[2:]) + case "coordinator": + err = coordinatorCommand(ctx, os.Args[2:]) + case "agent": + err = agentCommand(ctx, os.Args[2:]) + case "merge": + err = mergeCommand(os.Args[2:]) + case "collect": + err = collectCommand(ctx, os.Args[2:]) + default: + usage() + os.Exit(2) + } + if err != nil { + log.Fatal(err) + } +} + +func prepareCommand(ctx context.Context, args []string) error { + fs := flag.NewFlagSet("prepare", flag.ContinueOnError) + configPath := fs.String("config", "loadtest/im-pressure/config.example.json", "JSON configuration path") + manifestPath := fs.String("manifest", "loadtest/im-pressure/manifest.json", "manifest output path") + if err := fs.Parse(args); err != nil { + return err + } + cfg, err := loadConfig(*configPath) + if err != nil { + return err + } + return runPrepare(ctx, cfg, *manifestPath) +} + +func verifyCommand(ctx context.Context, args []string) error { + fs := flag.NewFlagSet("verify", flag.ContinueOnError) + configPath := fs.String("config", "loadtest/im-pressure/config.example.json", "JSON configuration path") + manifestPath := fs.String("manifest", "loadtest/im-pressure/manifest.json", "manifest path") + if err := fs.Parse(args); err != nil { + return err + } + cfg, err := loadConfig(*configPath) + if err != nil { + return err + } + manifest, err := loadManifest(*manifestPath) + if err != nil { + return err + } + apiKey, err := cfg.apiKey() + if err != nil { + return err + } + return runVerify(ctx, cfg, manifest, newAPIClient(cfg, apiKey)) +} + +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") + 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") + if err := fs.Parse(args); err != nil { + return err + } + return runCoordinator(ctx, *listen, *runID, *scenario, *output, *expected, *delay) +} + +func agentCommand(ctx context.Context, args []string) error { + fs := flag.NewFlagSet("agent", flag.ContinueOnError) + configPath := fs.String("config", "loadtest/im-pressure/config.example.json", "JSON configuration path") + manifestPath := fs.String("manifest", "loadtest/im-pressure/manifest.json", "manifest path") + machine := fs.String("machine", "", "a or b") + coordinatorURL := fs.String("coordinator", "", "coordinator base URL") + output := fs.String("output", "loadtest/im-pressure/results/agent-summary.json", "agent result path") + metricsListen := fs.String("metrics-listen", ":9105", "Prometheus listen address; empty disables it") + if err := fs.Parse(args); err != nil { + return err + } + if *coordinatorURL == "" { + return fmt.Errorf("--coordinator is required") + } + cfg, err := loadConfig(*configPath) + if err != nil { + return err + } + manifest, err := loadManifest(*manifestPath) + if err != nil { + return err + } + return runAgent(ctx, cfg, manifest, *machine, *coordinatorURL, *output, *metricsListen) +} + +func mergeCommand(args []string) error { + fs := flag.NewFlagSet("merge", flag.ContinueOnError) + a := fs.String("a", "loadtest/im-pressure/results/agent-a-summary.json", "machine A summary") + b := fs.String("b", "loadtest/im-pressure/results/agent-b-summary.json", "machine B summary") + output := fs.String("output", "loadtest/im-pressure/results/report.json", "combined report") + if err := fs.Parse(args); err != nil { + return err + } + return runMerge(*a, *b, *output) +} + +func collectCommand(ctx context.Context, args []string) error { + fs := flag.NewFlagSet("collect", flag.ContinueOnError) + var urls namedURLs + fs.Var(&urls, "url", "repeatable name=https://host/metrics source") + output := fs.String("output", "loadtest/im-pressure/results/metrics.jsonl", "snapshot output") + interval := fs.Duration("interval", 5*time.Second, "collection interval") + duration := fs.Duration("duration", 10*time.Minute, "collection duration") + if err := fs.Parse(args); err != nil { + return err + } + return runCollector(ctx, urls, *output, *interval, *duration) +} + +func usage() { + fmt.Fprintln(os.Stderr, "usage: im-pressure [flags]") +} diff --git a/loadtest/im-pressure/merge.go b/loadtest/im-pressure/merge.go new file mode 100644 index 0000000..92cc683 --- /dev/null +++ b/loadtest/im-pressure/merge.go @@ -0,0 +1,114 @@ +package main + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" + "sort" + "time" +) + +type CombinedReport struct { + RunID string `json:"run_id"` + Scenario string `json:"scenario"` + GeneratedAt time.Time `json:"generated_at"` + Agents []AgentSummary `json:"agents"` + ReadyConnections int `json:"ready_connections"` + PublishAttempted int64 `json:"publish_attempted"` + PublishAccepted int64 `json:"publish_accepted"` + PublishRejected int64 `json:"publish_rejected"` + ExpectedDeliveries int64 `json:"expected_deliveries"` + ReceivedDeliveries int64 `json:"received_deliveries"` + RealtimeLost int64 `json:"realtime_lost"` + RealtimeLostStored int64 `json:"realtime_lost_but_persisted"` + ArrivalRate float64 `json:"arrival_rate"` + LossRate float64 `json:"loss_rate"` + Duplicates int64 `json:"duplicates"` + OutOfOrder int64 `json:"out_of_order"` + Disconnects int64 `json:"disconnects"` + HistoryChecked int64 `json:"history_checked"` + HistoryFound int64 `json:"history_found"` + PersistenceLost int64 `json:"persistence_lost"` + Missing []MissingDelivery `json:"missing,omitempty"` + Passed bool `json:"passed"` + FailureReasons []string `json:"failure_reasons,omitempty"` +} + +func runMerge(aPath, bPath, output string) error { + a, err := readAgentSummary(aPath) + if err != nil { + return err + } + b, err := readAgentSummary(bPath) + if err != nil { + return err + } + if a.RunID != b.RunID || a.Scenario != b.Scenario { + return fmt.Errorf("agent summaries do not belong to the same run") + } + report := CombinedReport{RunID: a.RunID, Scenario: a.Scenario, GeneratedAt: time.Now().UTC(), Agents: []AgentSummary{a, b}, Passed: true} + for _, summary := range report.Agents { + report.ReadyConnections += summary.ReadyConnections + report.PublishAttempted += summary.PublishAttempted + report.PublishAccepted += summary.PublishAccepted + report.PublishRejected += summary.PublishRejected + report.ExpectedDeliveries += summary.ExpectedPeer + summary.ExpectedGroup + report.ReceivedDeliveries += summary.ReceivedPeer + summary.ReceivedGroup + report.RealtimeLost += summary.RealtimeLost + report.RealtimeLostStored += summary.RealtimeLostStored + report.Duplicates += summary.Duplicates + report.OutOfOrder += summary.OutOfOrder + report.Disconnects += summary.Disconnects + report.HistoryChecked += summary.HistoryChecked + report.HistoryFound += summary.HistoryFound + report.PersistenceLost += summary.PersistenceLost + report.Missing = append(report.Missing, summary.Missing...) + if len(summary.Errors) > 0 { + report.FailureReasons = append(report.FailureReasons, fmt.Sprintf("agent %s errors=%d", summary.Machine, len(summary.Errors))) + } + } + if report.ExpectedDeliveries > 0 { + report.ArrivalRate = float64(report.ReceivedDeliveries) / float64(report.ExpectedDeliveries) + report.LossRate = float64(report.RealtimeLost) / float64(report.ExpectedDeliveries) + } + if report.ReadyConnections != 2000 { + report.FailureReasons = append(report.FailureReasons, fmt.Sprintf("ready connections=%d expected=2000", report.ReadyConnections)) + } + if report.PublishRejected > 0 || report.RealtimeLost > 0 || report.PersistenceLost > 0 || report.Duplicates > 0 || report.OutOfOrder > 0 || report.Disconnects > 0 { + report.FailureReasons = append(report.FailureReasons, "message correctness gate failed") + } + report.Passed = len(report.FailureReasons) == 0 + sort.Slice(report.Missing, func(i, j int) bool { + if report.Missing[i].ClientMid == report.Missing[j].ClientMid { + return report.Missing[i].ReceiverIndex < report.Missing[j].ReceiverIndex + } + return report.Missing[i].ClientMid < report.Missing[j].ClientMid + }) + return writeJSONFile(output, report) +} + +func readAgentSummary(path string) (AgentSummary, error) { + var summary AgentSummary + raw, err := os.ReadFile(path) + if err != nil { + return summary, err + } + if err = json.Unmarshal(raw, &summary); err != nil { + return summary, err + } + return summary, nil +} + +func writeJSONFile(path string, value any) error { + raw, err := json.MarshalIndent(value, "", " ") + if err != nil { + return err + } + if dir := filepath.Dir(path); dir != "." { + if err = os.MkdirAll(dir, 0o750); err != nil { + return err + } + } + return os.WriteFile(path, append(raw, '\n'), 0o600) +} diff --git a/loadtest/im-pressure/merge_test.go b/loadtest/im-pressure/merge_test.go new file mode 100644 index 0000000..6e3e00d --- /dev/null +++ b/loadtest/im-pressure/merge_test.go @@ -0,0 +1,49 @@ +package main + +import ( + "encoding/json" + "os" + "path/filepath" + "testing" +) + +func TestRunMergeComputesArrivalAndPersistenceGate(t *testing.T) { + dir := t.TempDir() + a := AgentSummary{RunID: "run1", Scenario: "p2p", Machine: "a", ReadyConnections: 1000, + PublishAttempted: 10, PublishAccepted: 10, ExpectedPeer: 10, ReceivedPeer: 9, + RealtimeLost: 1, HistoryChecked: 10, HistoryFound: 10} + b := AgentSummary{RunID: "run1", Scenario: "p2p", Machine: "b", ReadyConnections: 1000, + PublishAttempted: 10, PublishAccepted: 10, ExpectedPeer: 10, ReceivedPeer: 10, + HistoryChecked: 10, HistoryFound: 9, PersistenceLost: 1} + aPath, bPath, reportPath := filepath.Join(dir, "a.json"), filepath.Join(dir, "b.json"), filepath.Join(dir, "report.json") + writeSummaryForTest(t, aPath, a) + writeSummaryForTest(t, bPath, b) + if err := runMerge(aPath, bPath, reportPath); err != nil { + t.Fatal(err) + } + var report CombinedReport + raw, err := os.ReadFile(reportPath) + if err != nil { + t.Fatal(err) + } + if err = json.Unmarshal(raw, &report); err != nil { + t.Fatal(err) + } + if report.ExpectedDeliveries != 20 || report.ReceivedDeliveries != 19 || report.ArrivalRate != 0.95 || report.LossRate != 0.05 { + t.Fatalf("unexpected delivery totals: %+v", report) + } + if report.Passed || report.PersistenceLost != 1 { + t.Fatalf("correctness gate did not fail: %+v", report) + } +} + +func writeSummaryForTest(t *testing.T, path string, summary AgentSummary) { + t.Helper() + raw, err := json.Marshal(summary) + if err != nil { + t.Fatal(err) + } + if err = os.WriteFile(path, raw, 0o600); err != nil { + t.Fatal(err) + } +} diff --git a/loadtest/im-pressure/model.go b/loadtest/im-pressure/model.go new file mode 100644 index 0000000..1f841e2 --- /dev/null +++ b/loadtest/im-pressure/model.go @@ -0,0 +1,294 @@ +package main + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "os" + "sort" + "strings" + "time" +) + +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"` + InsecureTLS bool `json:"insecure_tls"` + PrepareWorkers int `json:"prepare_workers"` + ConnectRamp string `json:"connect_ramp"` + CommandTimeout string `json:"command_timeout"` + P2PMessages int `json:"p2p_messages_per_user"` + P2PInterval string `json:"p2p_interval"` + DrainP2P string `json:"drain_p2p"` + DrainGroup string `json:"drain_group"` +} + +func loadConfig(path string) (*Config, error) { + raw, err := os.ReadFile(path) + if err != nil { + return nil, err + } + var cfg Config + if err = json.Unmarshal(raw, &cfg); err != nil { + return nil, fmt.Errorf("decode config: %w", err) + } + if cfg.Tenant == "" { + cfg.Tenant = "006" + } + if cfg.UsersPerMachine <= 0 { + cfg.UsersPerMachine = 1000 + } + if cfg.PrepareWorkers <= 0 { + cfg.PrepareWorkers = 20 + } + if cfg.ConnectRamp == "" { + cfg.ConnectRamp = "20s" + } + if cfg.CommandTimeout == "" { + cfg.CommandTimeout = "20s" + } + if cfg.P2PMessages <= 0 { + cfg.P2PMessages = 10 + } + if cfg.P2PInterval == "" { + cfg.P2PInterval = "1s" + } + if cfg.DrainP2P == "" { + cfg.DrainP2P = "30s" + } + if cfg.DrainGroup == "" { + cfg.DrainGroup = "60s" + } + if strings.TrimSpace(cfg.HTTPBase) == "" { + return nil, errors.New("http_base is required") + } + for _, machine := range []string{"a", "b"} { + if len(cfg.Endpoints[machine]) == 0 { + return nil, fmt.Errorf("at least one endpoint is required for machine %s", machine) + } + } + return &cfg, nil +} + +func (c *Config) apiKey() (string, error) { + name := c.APIKeyEnv + if name == "" { + name = "IM_API_KEY" + } + value := strings.TrimSpace(os.Getenv(name)) + if value == "" { + return "", fmt.Errorf("environment variable %s is required", name) + } + return value, nil +} + +func (c *Config) password() (string, error) { + name := c.PasswordEnv + if name == "" { + name = "IM_LOAD_PASSWORD" + } + value := os.Getenv(name) + if value == "" { + return "", fmt.Errorf("environment variable %s is required", name) + } + return value, nil +} + +func configDuration(value string) (time.Duration, error) { + d, err := time.ParseDuration(value) + if err != nil || d <= 0 { + return 0, fmt.Errorf("invalid positive duration %q", value) + } + return d, nil +} + +type Manifest struct { + Version int `json:"version"` + Tenant string `json:"tenant"` + GroupTopic string `json:"group_topic,omitempty"` + CreatedAt time.Time `json:"created_at"` + Accounts []AccountRecord `json:"accounts"` +} + +type AccountRecord struct { + Machine string `json:"machine"` + Index int `json:"index"` + Username string `json:"username"` + UserID string `json:"user_id"` + Token string `json:"token"` + TokenExpires time.Time `json:"token_expires,omitempty"` + P2PTopic string `json:"p2p_topic,omitempty"` + Endpoint string `json:"endpoint,omitempty"` +} + +func newManifest(cfg *Config) *Manifest { + m := &Manifest{Version: manifestVersion, Tenant: cfg.Tenant, CreatedAt: time.Now().UTC()} + for _, machine := range []string{"a", "b"} { + for i := 1; i <= cfg.UsersPerMachine; i++ { + m.Accounts = append(m.Accounts, AccountRecord{ + Machine: machine, + Index: i, + Username: fmt.Sprintf("lt%s%04d", machine, i), + Endpoint: cfg.Endpoints[machine][(i-1)%len(cfg.Endpoints[machine])], + }) + } + } + return m +} + +func applyConfigEndpoints(cfg *Config, manifest *Manifest) error { + if manifest.Tenant != cfg.Tenant { + return fmt.Errorf("manifest tenant=%s config tenant=%s", manifest.Tenant, cfg.Tenant) + } + if len(manifest.Accounts) != cfg.UsersPerMachine*2 { + return fmt.Errorf("manifest accounts=%d expected=%d", len(manifest.Accounts), cfg.UsersPerMachine*2) + } + for i := range manifest.Accounts { + account := &manifest.Accounts[i] + endpoints := cfg.Endpoints[account.Machine] + if account.Index <= 0 || len(endpoints) == 0 { + return fmt.Errorf("invalid endpoint assignment for %s", accountKey(account.Machine, account.Index)) + } + account.Endpoint = endpoints[(account.Index-1)%len(endpoints)] + } + return nil +} + +func loadManifest(path string) (*Manifest, error) { + raw, err := os.ReadFile(path) + if err != nil { + return nil, err + } + var manifest Manifest + if err = json.Unmarshal(raw, &manifest); err != nil { + return nil, err + } + if manifest.Version != manifestVersion { + return nil, fmt.Errorf("unsupported manifest version %d", manifest.Version) + } + return &manifest, manifest.validate() +} + +func (m *Manifest) validate() error { + seen := make(map[string]bool, len(m.Accounts)) + for _, account := range m.Accounts { + key := accountKey(account.Machine, account.Index) + if account.Machine != "a" && account.Machine != "b" { + return fmt.Errorf("invalid machine %q", account.Machine) + } + if account.Index <= 0 || account.Username == "" || seen[key] { + return fmt.Errorf("invalid or duplicate account %s", key) + } + seen[key] = true + } + return nil +} + +func saveManifest(path string, manifest *Manifest) error { + if err := manifest.validate(); err != nil { + return err + } + raw, err := json.MarshalIndent(manifest, "", " ") + if err != nil { + return err + } + tmp := path + ".tmp" + if err = os.WriteFile(tmp, append(raw, '\n'), 0o600); err != nil { + return err + } + return os.Rename(tmp, path) +} + +func accountKey(machine string, index int) string { + return strings.ToLower(machine) + fmt.Sprintf("-%04d", index) +} + +func accountsForMachine(manifest *Manifest, machine string) []AccountRecord { + var out []AccountRecord + for _, account := range manifest.Accounts { + if account.Machine == machine { + out = append(out, account) + } + } + sort.Slice(out, func(i, j int) bool { return out[i].Index < out[j].Index }) + return out +} + +func manifestAccountMap(manifest *Manifest) map[string]*AccountRecord { + out := make(map[string]*AccountRecord, len(manifest.Accounts)) + for i := range manifest.Accounts { + account := &manifest.Accounts[i] + out[accountKey(account.Machine, account.Index)] = account + } + return out +} + +func topicHash(topic string) string { + sum := sha256.Sum256([]byte(topic)) + return hex.EncodeToString(sum[:6]) +} + +type AcceptedMessage struct { + RunID string `json:"run_id"` + Scenario string `json:"scenario"` + Kind string `json:"kind"` + ClientMid string `json:"client_mid"` + Topic string `json:"topic"` + Seq int64 `json:"seq"` + SenderMachine string `json:"sender_machine"` + SenderIndex int `json:"sender_index"` + PeerMachine string `json:"peer_machine,omitempty"` + PeerIndex int `json:"peer_index,omitempty"` + SentAt time.Time `json:"sent_at"` + AckAt time.Time `json:"ack_at"` +} + +type AgentSummary struct { + RunID string `json:"run_id"` + Scenario string `json:"scenario"` + Machine string `json:"machine"` + StartedAt time.Time `json:"started_at"` + LoadEndedAt time.Time `json:"load_ended_at"` + FinishedAt time.Time `json:"finished_at"` + ReadyConnections int `json:"ready_connections"` + PublishAttempted int64 `json:"publish_attempted"` + PublishAccepted int64 `json:"publish_accepted"` + PublishRejected int64 `json:"publish_rejected"` + ExpectedPeer int64 `json:"expected_peer"` + ReceivedPeer int64 `json:"received_peer"` + ExpectedEcho int64 `json:"expected_echo"` + ReceivedEcho int64 `json:"received_echo"` + ExpectedGroup int64 `json:"expected_group"` + ReceivedGroup int64 `json:"received_group"` + RealtimeLost int64 `json:"realtime_lost"` + RealtimeLostStored int64 `json:"realtime_lost_but_persisted"` + Duplicates int64 `json:"duplicates"` + OutOfOrder int64 `json:"out_of_order"` + Disconnects int64 `json:"disconnects"` + UnexpectedMessages int64 `json:"unexpected_messages"` + HistoryChecked int64 `json:"history_checked"` + HistoryFound int64 `json:"history_found"` + PersistenceLost int64 `json:"persistence_lost"` + AckLatency HistogramSnapshot `json:"ack_latency"` + DeliveryLatency HistogramSnapshot `json:"delivery_latency"` + Errors []string `json:"errors,omitempty"` + Missing []MissingDelivery `json:"missing,omitempty"` + PersistenceMissing []string `json:"persistence_missing,omitempty"` + AcceptedLocal []AcceptedMessage `json:"accepted_local,omitempty"` +} + +type MissingDelivery struct { + ClientMid string `json:"client_mid"` + Kind string `json:"kind"` + TopicHash string `json:"topic_hash"` + Seq int64 `json:"seq"` + ReceiverIndex int `json:"receiver_index"` +} diff --git a/loadtest/im-pressure/model_test.go b/loadtest/im-pressure/model_test.go new file mode 100644 index 0000000..94b38c4 --- /dev/null +++ b/loadtest/im-pressure/model_test.go @@ -0,0 +1,45 @@ +package main + +import "testing" + +func TestNewManifestDistributesUsersAndEndpoints(t *testing.T) { + cfg := &Config{Tenant: "006", UsersPerMachine: 3, Endpoints: map[string][]string{ + "a": {"ws://a1", "ws://a2"}, "b": {"ws://b1"}, + }} + manifest := newManifest(cfg) + if len(manifest.Accounts) != 6 { + t.Fatalf("accounts=%d", len(manifest.Accounts)) + } + if got := accountsForMachine(manifest, "a"); got[0].Username != "lta0001" || got[1].Endpoint != "ws://a2" || got[2].Endpoint != "ws://a1" { + t.Fatalf("unexpected A accounts: %+v", got) + } + if err := manifest.validate(); err != nil { + t.Fatal(err) + } +} + +func TestManifestRejectsDuplicateAccount(t *testing.T) { + manifest := &Manifest{Version: manifestVersion, Accounts: []AccountRecord{ + {Machine: "a", Index: 1, Username: "lta0001"}, + {Machine: "a", Index: 1, Username: "duplicate"}, + }} + if err := manifest.validate(); err == nil { + t.Fatal("expected duplicate account validation error") + } +} + +func TestApplyConfigEndpointsRefreshesManifest(t *testing.T) { + cfg := &Config{Tenant: "006", UsersPerMachine: 1, Endpoints: map[string][]string{ + "a": {"ws://new-a"}, "b": {"ws://new-b"}, + }} + manifest := &Manifest{Tenant: "006", Accounts: []AccountRecord{ + {Machine: "a", Index: 1, Username: "lta0001", Endpoint: "ws://old"}, + {Machine: "b", Index: 1, Username: "ltb0001", Endpoint: "ws://old"}, + }} + if err := applyConfigEndpoints(cfg, manifest); err != nil { + t.Fatal(err) + } + if manifest.Accounts[0].Endpoint != "ws://new-a" || manifest.Accounts[1].Endpoint != "ws://new-b" { + t.Fatalf("endpoints were not refreshed: %+v", manifest.Accounts) + } +} diff --git a/loadtest/im-pressure/prepare.go b/loadtest/im-pressure/prepare.go new file mode 100644 index 0000000..580c9ed --- /dev/null +++ b/loadtest/im-pressure/prepare.go @@ -0,0 +1,253 @@ +package main + +import ( + "context" + "errors" + "fmt" + "log" + "net/http" + "sort" + "sync" + "sync/atomic" +) + +func runPrepare(ctx context.Context, cfg *Config, manifestPath string) error { + apiKey, err := cfg.apiKey() + if err != nil { + return err + } + password, err := cfg.password() + if err != nil { + return err + } + manifest := newManifest(cfg) + if existing, loadErr := loadManifest(manifestPath); loadErr == nil { + manifest = existing + log.Printf("resuming manifest with %d accounts", len(manifest.Accounts)) + } + if err = applyConfigEndpoints(cfg, manifest); err != nil { + return err + } + if err = prepareAccounts(ctx, cfg, manifest, apiKey, password); err != nil { + return err + } + if err = saveManifest(manifestPath, manifest); err != nil { + return err + } + api := newAPIClient(cfg, apiKey) + if err = prepareFriendships(ctx, cfg, manifest, api); err != nil { + return err + } + if err = saveManifest(manifestPath, manifest); err != nil { + return err + } + if err = prepareGroup(ctx, cfg, manifest, api); err != nil { + return err + } + if err = saveManifest(manifestPath, manifest); err != nil { + return err + } + return runVerify(ctx, cfg, manifest, api) +} + +func prepareAccounts(ctx context.Context, cfg *Config, manifest *Manifest, apiKey, password string) error { + var created atomic.Int64 + var existing atomic.Int64 + var mu sync.Mutex + err := parallelAccounts(ctx, manifest.Accounts, cfg.PrepareWorkers, func(account AccountRecord) error { + client, wasCreated, err := connectForPrepare(ctx, cfg, account, apiKey, password) + if err != nil { + return fmt.Errorf("account %s: %w", account.Username, err) + } + updated := client.Account() + client.Close() + mu.Lock() + defer mu.Unlock() + for i := range manifest.Accounts { + if manifest.Accounts[i].Machine == account.Machine && manifest.Accounts[i].Index == account.Index { + manifest.Accounts[i].UserID = updated.UserID + manifest.Accounts[i].Token = updated.Token + manifest.Accounts[i].TokenExpires = updated.TokenExpires + break + } + } + if wasCreated { + created.Add(1) + } else { + existing.Add(1) + } + return nil + }) + if err == nil { + log.Printf("accounts ready=%d created=%d existing=%d", len(manifest.Accounts), created.Load(), existing.Load()) + } + return err +} + +func prepareFriendships(ctx context.Context, cfg *Config, manifest *Manifest, api *apiClient) error { + accounts := manifestAccountMap(manifest) + pairs := make([]AccountRecord, 0, cfg.UsersPerMachine) + for i := 1; i <= cfg.UsersPerMachine; i++ { + pairs = append(pairs, *accounts[accountKey("a", i)]) + } + var mu sync.Mutex + err := parallelAccounts(ctx, pairs, cfg.PrepareWorkers, func(a AccountRecord) error { + b := *accounts[accountKey("b", a.Index)] + if a.UserID == "" || b.UserID == "" { + return fmt.Errorf("pair %d has missing user ID", a.Index) + } + request, requestErr := api.createFriendRequest(ctx, a, b) + if requestErr == nil && request.Status == "pending" { + if _, requestErr = api.acceptFriendRequest(ctx, b, request.RequestID); requestErr != nil { + return fmt.Errorf("accept pair %d: %w", a.Index, requestErr) + } + } else if requestErr != nil { + var apiErr *apiError + if !errors.As(requestErr, &apiErr) || apiErr.Status != http.StatusConflict { + return fmt.Errorf("friend pair %d: %w", a.Index, requestErr) + } + pendingID, findErr := api.findPendingFriendRequest(ctx, b, a.UserID) + if findErr != nil { + return fmt.Errorf("find pending friend pair %d: %w", a.Index, findErr) + } + if pendingID != "" { + if _, acceptErr := api.acceptFriendRequest(ctx, b, pendingID); acceptErr != nil { + return fmt.Errorf("resume friend pair %d: %w", a.Index, acceptErr) + } + } + } + conversation, err := api.ensureP2P(ctx, a, b) + if err != nil { + return fmt.Errorf("ensure P2P pair %d: %w", a.Index, err) + } + if conversation.Relationship != "friend" || conversation.Topic == "" { + return fmt.Errorf("pair %d relationship=%s topic=%q", a.Index, conversation.Relationship, conversation.Topic) + } + mu.Lock() + accounts[accountKey("a", a.Index)].P2PTopic = conversation.Topic + accounts[accountKey("b", a.Index)].P2PTopic = conversation.Topic + mu.Unlock() + return nil + }) + if err == nil { + log.Printf("friendships ready=%d", cfg.UsersPerMachine) + } + return err +} + +func prepareGroup(ctx context.Context, cfg *Config, manifest *Manifest, api *apiClient) error { + if err := parallelAccounts(ctx, manifest.Accounts, cfg.PrepareWorkers, func(account AccountRecord) error { + return api.disableGroupInviteVerification(ctx, account) + }); err != nil { + return fmt.Errorf("prepare group invite privacy: %w", err) + } + accounts := manifestAccountMap(manifest) + owner := *accounts[accountKey("a", 1)] + if manifest.GroupTopic == "" { + group, err := api.createGroup(ctx, owner) + if err != nil { + return err + } + manifest.GroupTopic = group.Topic + log.Printf("created load-test group topic_hash=%s", topicHash(group.Topic)) + } + memberIDs := make([]string, 0, len(manifest.Accounts)-1) + for _, account := range manifest.Accounts { + if account.UserID != owner.UserID { + memberIDs = append(memberIDs, account.UserID) + } + } + sort.Strings(memberIDs) + for offset, batch := 0, 1; offset < len(memberIDs); offset, batch = offset+100, batch+1 { + end := offset + 100 + if end > len(memberIDs) { + end = len(memberIDs) + } + response, err := api.inviteGroup(ctx, owner, manifest.GroupTopic, batch, memberIDs[offset:end]) + if err != nil { + return fmt.Errorf("invite group batch %d: %w", batch, err) + } + for _, item := range response.Items { + if item.Status != "joined" && item.Status != "already_member" { + return fmt.Errorf("invite user %s status=%s reason=%s", item.UserID, item.Status, item.Reason) + } + } + } + count, err := api.countGroupMembers(ctx, owner, manifest.GroupTopic) + if err != nil { + return err + } + if count != len(manifest.Accounts) { + return fmt.Errorf("group has %d members, expected %d", count, len(manifest.Accounts)) + } + log.Printf("group ready members=%d topic_hash=%s", count, topicHash(manifest.GroupTopic)) + return nil +} + +func runVerify(ctx context.Context, cfg *Config, manifest *Manifest, api *apiClient) error { + if len(manifest.Accounts) != cfg.UsersPerMachine*2 { + return fmt.Errorf("manifest accounts=%d expected=%d", len(manifest.Accounts), cfg.UsersPerMachine*2) + } + for _, account := range manifest.Accounts { + if account.UserID == "" || account.Token == "" || account.P2PTopic == "" { + return fmt.Errorf("account %s is incomplete", account.Username) + } + } + if manifest.GroupTopic == "" { + return errors.New("manifest group topic is empty") + } + owner := manifestAccountMap(manifest)[accountKey("a", 1)] + count, err := api.countGroupMembers(ctx, *owner, manifest.GroupTopic) + if err != nil { + return err + } + if count != len(manifest.Accounts) { + return fmt.Errorf("group member count=%d expected=%d", count, len(manifest.Accounts)) + } + log.Printf("verify passed accounts=%d pairs=%d group_members=%d", len(manifest.Accounts), cfg.UsersPerMachine, count) + return nil +} + +func parallelAccounts(ctx context.Context, accounts []AccountRecord, workers int, fn func(AccountRecord) error) error { + if workers <= 0 { + workers = 1 + } + workCtx, cancel := context.WithCancel(ctx) + defer cancel() + jobs := make(chan AccountRecord) + errCh := make(chan error, 1) + var wg sync.WaitGroup + for i := 0; i < workers; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for account := range jobs { + if err := fn(account); err != nil { + select { + case errCh <- err: + default: + } + cancel() + return + } + } + }() + } + go func() { + defer close(jobs) + for _, account := range accounts { + select { + case jobs <- account: + case <-workCtx.Done(): + return + } + } + }() + wg.Wait() + select { + case err := <-errCh: + return err + default: + return ctx.Err() + } +} diff --git a/loadtest/im-pressure/recorder.go b/loadtest/im-pressure/recorder.go new file mode 100644 index 0000000..571cf88 --- /dev/null +++ b/loadtest/im-pressure/recorder.go @@ -0,0 +1,229 @@ +package main + +import ( + "encoding/json" + "fmt" + "math/bits" + "strings" + "sync" + "sync/atomic" + "time" +) + +type observation struct { + topic string + seq int64 + bits []uint64 +} + +type deliveryRecorder struct { + machine string + localUsers int + latency *DurationHistogram + disconnects atomic.Int64 + received atomic.Int64 + runID atomic.Value + historyMode atomic.Bool + + mu sync.Mutex + observed map[string]*observation + lastSeq []map[string]int64 + duplicates int64 + outOfOrder int64 + malformed int64 + history map[string]bool +} + +func newDeliveryRecorder(runID, machine string, localUsers int) *deliveryRecorder { + lastSeq := make([]map[string]int64, localUsers) + for i := range lastSeq { + lastSeq[i] = make(map[string]int64) + } + recorder := &deliveryRecorder{ + machine: machine, localUsers: localUsers, latency: NewDurationHistogram(), + observed: make(map[string]*observation), lastSeq: lastSeq, history: make(map[string]bool), + } + recorder.runID.Store(runID) + return recorder +} + +func (r *deliveryRecorder) setRunID(runID string) { r.runID.Store(runID) } + +func (r *deliveryRecorder) onData(account AccountRecord, data dataPacket) { + runID, _ := r.runID.Load().(string) + mid, _ := data.Head["x-im-client-mid"].(string) + if mid == "" || !strings.HasPrefix(mid, runID+"-") { + return + } + var content struct { + RunID string `json:"run_id"` + SentAt int64 `json:"sent_at_unix_nano"` + } + if json.Unmarshal(data.Content, &content) != nil || content.RunID != runID { + r.mu.Lock() + r.malformed++ + r.mu.Unlock() + return + } + if r.historyMode.Load() { + r.mu.Lock() + r.history[mid] = true + r.mu.Unlock() + return + } + index := account.Index - 1 + if index < 0 || index >= r.localUsers { + return + } + r.mu.Lock() + defer r.mu.Unlock() + obs := r.observed[mid] + if obs == nil { + obs = &observation{topic: data.Topic, seq: data.Seq, bits: make([]uint64, (r.localUsers+63)/64)} + r.observed[mid] = obs + } + word, bit := index/64, uint(index%64) + mask := uint64(1) << bit + if obs.bits[word]&mask != 0 { + r.duplicates++ + return + } + obs.bits[word] |= mask + r.received.Add(1) + if previous := r.lastSeq[index][data.Topic]; previous > 0 && data.Seq < previous { + r.outOfOrder++ + } + if data.Seq > r.lastSeq[index][data.Topic] { + r.lastSeq[index][data.Topic] = data.Seq + } + if content.SentAt > 0 { + r.latency.Observe(time.Since(time.Unix(0, content.SentAt))) + } +} + +func (r *deliveryRecorder) metricCounts() (received, duplicates, outOfOrder, disconnects int64) { + r.mu.Lock() + duplicates = r.duplicates + outOfOrder = r.outOfOrder + r.mu.Unlock() + return r.received.Load(), duplicates, outOfOrder, r.disconnects.Load() +} + +func (r *deliveryRecorder) beginHistory() { r.historyMode.Store(true) } + +func (r *deliveryRecorder) historyFound(mid string) bool { + r.mu.Lock() + defer r.mu.Unlock() + return r.history[mid] +} + +func (r *deliveryRecorder) onDisconnect(_ AccountRecord, _ error) { + r.disconnects.Add(1) +} + +func (r *deliveryRecorder) summarize(base AgentSummary, ledger []AcceptedMessage) AgentSummary { + base.DeliveryLatency = r.latency.Snapshot() + base.Disconnects = r.disconnects.Load() + runID, _ := r.runID.Load().(string) + accepted := make(map[string]AcceptedMessage, len(ledger)) + const missingLimit = 100000 + + r.mu.Lock() + defer r.mu.Unlock() + base.Duplicates = r.duplicates + base.OutOfOrder = r.outOfOrder + if r.malformed > 0 { + base.Errors = append(base.Errors, fmt.Sprintf("malformed run messages=%d", r.malformed)) + } + for _, message := range ledger { + if message.RunID != runID { + continue + } + accepted[message.ClientMid] = message + obs := r.observed[message.ClientMid] + switch message.Kind { + case "group": + base.ExpectedGroup += int64(r.localUsers) + for index := 1; index <= r.localUsers; index++ { + if observationHas(obs, index) { + base.ReceivedGroup++ + } else if len(base.Missing) < missingLimit { + base.Missing = append(base.Missing, missingFor(message, index)) + } + } + case "p2p": + if message.PeerMachine == r.machine && message.PeerIndex > 0 { + base.ExpectedPeer++ + if observationHas(obs, message.PeerIndex) { + base.ReceivedPeer++ + } else if len(base.Missing) < missingLimit { + base.Missing = append(base.Missing, missingFor(message, message.PeerIndex)) + } + } + if message.SenderMachine == r.machine && message.SenderIndex > 0 { + base.ExpectedEcho++ + if observationHas(obs, message.SenderIndex) { + base.ReceivedEcho++ + } + } + } + } + for mid, obs := range r.observed { + if _, ok := accepted[mid]; !ok { + base.UnexpectedMessages += bitsetCount(obs.bits) + } + } + base.RealtimeLost = (base.ExpectedPeer - base.ReceivedPeer) + (base.ExpectedGroup - base.ReceivedGroup) + if len(base.Missing) == missingLimit { + base.Errors = append(base.Errors, "missing delivery details truncated at 100000") + } + return base +} + +func (r *deliveryRecorder) countRealtimeLostStored(ledger []AcceptedMessage) int64 { + runID, _ := r.runID.Load().(string) + r.mu.Lock() + defer r.mu.Unlock() + var count int64 + for _, message := range ledger { + if message.RunID != runID || !r.history[message.ClientMid] { + continue + } + obs := r.observed[message.ClientMid] + switch message.Kind { + case "group": + for index := 1; index <= r.localUsers; index++ { + if !observationHas(obs, index) { + count++ + } + } + case "p2p": + if message.PeerMachine == r.machine && message.PeerIndex > 0 && !observationHas(obs, message.PeerIndex) { + count++ + } + } + } + return count +} + +func observationHas(obs *observation, receiverIndex int) bool { + if obs == nil || receiverIndex <= 0 { + return false + } + index := receiverIndex - 1 + word, bit := index/64, uint(index%64) + return word < len(obs.bits) && obs.bits[word]&(uint64(1)< 0 { + _ = json.Unmarshal(raw, &c.account.UserID) + } + if raw := ctrl.Params["token"]; len(raw) > 0 { + _ = json.Unmarshal(raw, &c.account.Token) + } + if raw := ctrl.Params["expires"]; len(raw) > 0 { + var value string + if json.Unmarshal(raw, &value) == nil { + c.account.TokenExpires, _ = time.Parse(time.RFC3339Nano, value) + } + } +} + +func (c *WSClient) Account() AccountRecord { return c.account } + +func (c *WSClient) Subscribe(ctx context.Context, topic string) error { + if topic == "" { + return errors.New("empty topic") + } + id := c.nextID("sub") + _, err := c.command(ctx, "sub", map[string]any{"sub": map[string]any{ + "id": id, "topic": topic, "get": map[string]any{"what": "desc"}, + }}) + return err +} + +func (c *WSClient) GetData(ctx context.Context, topic string, since, before int64, limit int) error { + if topic == "" || since <= 0 || before <= since || limit <= 0 { + return errors.New("invalid get data range") + } + id := c.nextID("get-data") + _, err := c.command(ctx, "get", map[string]any{"get": map[string]any{ + "id": id, "topic": topic, "what": "data", + "data": map[string]any{"since": since, "before": before, "limit": limit}, + }}) + return err +} + +func (c *WSClient) Publish(ctx context.Context, topic, clientMid, runID, scenario string, messageIndex int) (AcceptedMessage, error) { + id := c.nextID("pub") + sentAt := time.Now().UTC() + ctrl, err := c.command(ctx, "pub", map[string]any{"pub": map[string]any{ + "id": id, "topic": topic, "noecho": false, + "head": map[string]any{"mime": "application/json", "x-im-schema": 1, "x-im-type": "text", "x-im-client-mid": clientMid}, + "content": map[string]any{"text": "loadtest", "run_id": runID, "scenario": scenario, + "sender_index": c.account.Index, "message_index": messageIndex, "sent_at_unix_nano": sentAt.UnixNano()}, + }}) + if err != nil { + return AcceptedMessage{}, err + } + seq, err := ctrlInt64(ctrl.Params["seq"]) + if err != nil || seq <= 0 { + return AcceptedMessage{}, fmt.Errorf("publish ACK has invalid seq: %w", err) + } + kind := "p2p" + if strings.HasPrefix(topic, "grp") { + kind = "group" + } + return AcceptedMessage{RunID: runID, Scenario: scenario, Kind: kind, ClientMid: clientMid, + Topic: topic, Seq: seq, SenderMachine: c.account.Machine, SenderIndex: c.account.Index, + SentAt: sentAt, AckAt: time.Now().UTC()}, nil +} + +func ctrlInt64(raw json.RawMessage) (int64, error) { + if len(raw) == 0 { + return 0, errors.New("missing value") + } + var number json.Number + if err := json.Unmarshal(raw, &number); err == nil { + return strconv.ParseInt(number.String(), 10, 64) + } + var text string + if err := json.Unmarshal(raw, &text); err != nil { + return 0, err + } + return strconv.ParseInt(text, 10, 64) +} + +func (c *WSClient) command(ctx context.Context, _ string, message map[string]any) (ctrlPacket, error) { + id := nestedMessageID(message) + if id == "" { + return ctrlPacket{}, errors.New("command id is required") + } + wait := make(chan ctrlPacket, 1) + c.mu.Lock() + c.waits[id] = wait + c.mu.Unlock() + c.sendMu.Lock() + err := c.conn.WriteJSON(message) + c.sendMu.Unlock() + if err != nil { + c.removeWait(id) + return ctrlPacket{}, err + } + timer := time.NewTimer(c.commandTimeout) + defer timer.Stop() + select { + case ctrl := <-wait: + if ctrl.Code < 200 || ctrl.Code >= 300 { + return ctrl, &ctrlError{Code: ctrl.Code, Text: ctrl.Text} + } + return ctrl, nil + case <-timer.C: + c.removeWait(id) + return ctrlPacket{}, fmt.Errorf("ctrl timeout id=%s", id) + case <-ctx.Done(): + c.removeWait(id) + return ctrlPacket{}, ctx.Err() + case <-c.done: + c.removeWait(id) + return ctrlPacket{}, errors.New("websocket closed") + } +} + +func nestedMessageID(message map[string]any) string { + for _, value := range message { + if body, ok := value.(map[string]any); ok { + if id, ok := body["id"].(string); ok { + return id + } + } + } + return "" +} + +func (c *WSClient) nextID(prefix string) string { + n := atomic.AddUint64(&c.seq, 1) + return fmt.Sprintf("%s-%s-%04d-%d", prefix, c.account.Machine, c.account.Index, n) +} + +func (c *WSClient) removeWait(id string) { + c.mu.Lock() + delete(c.waits, id) + c.mu.Unlock() +} + +func (c *WSClient) readLoop() { + defer close(c.done) + for { + var envelope map[string]json.RawMessage + if err := c.conn.ReadJSON(&envelope); err != nil { + if !c.closed.Load() && c.onDisconnect != nil { + c.onDisconnect(c.account, err) + } + return + } + if raw := envelope["ctrl"]; len(raw) > 0 { + var ctrl ctrlPacket + if json.Unmarshal(raw, &ctrl) == nil && ctrl.ID != "" { + c.mu.Lock() + wait := c.waits[ctrl.ID] + delete(c.waits, ctrl.ID) + c.mu.Unlock() + if wait != nil { + wait <- ctrl + } + } + continue + } + if raw := envelope["data"]; len(raw) > 0 && c.onData != nil { + var data dataPacket + if json.Unmarshal(raw, &data) == nil { + c.onData(c.account, data) + } + } + } +} + +func (c *WSClient) Close() { + if c == nil || !c.closed.CompareAndSwap(false, true) { + return + } + _ = c.conn.Close() +}