diff --git a/.gitea/workflows/deploy.yml b/.gitea/workflows/deploy.yml index 2dc9d36..1ecb563 100644 --- a/.gitea/workflows/deploy.yml +++ b/.gitea/workflows/deploy.yml @@ -62,11 +62,22 @@ jobs: kubectl rollout restart statefulset/im-server -n im-server-system kubectl rollout status statefulset/im-server -n im-server-system --timeout=300s - resources="$(kubectl get statefulset/im-server -n im-server-system -o \ - jsonpath='{.spec.template.spec.containers[?(@.name=="im-server")].resources}')" + echo "Waiting for ArgoCD reconciliation before validating effective settings" + sleep 60 + request_cpu="$(kubectl get statefulset/im-server -n im-server-system -o \ + jsonpath='{.spec.template.spec.containers[?(@.name=="im-server")].resources.requests.cpu}')" + request_memory="$(kubectl get statefulset/im-server -n im-server-system -o \ + jsonpath='{.spec.template.spec.containers[?(@.name=="im-server")].resources.requests.memory}')" + limit_cpu="$(kubectl get statefulset/im-server -n im-server-system -o \ + jsonpath='{.spec.template.spec.containers[?(@.name=="im-server")].resources.limits.cpu}')" + limit_memory="$(kubectl get statefulset/im-server -n im-server-system -o \ + jsonpath='{.spec.template.spec.containers[?(@.name=="im-server")].resources.limits.memory}')" liveness="$(kubectl get statefulset/im-server -n im-server-system -o \ jsonpath='{.spec.template.spec.containers[?(@.name=="im-server")].livenessProbe.httpGet.path}')" - echo "effective resources=${resources} liveness=${liveness}" - if [ "${liveness}" != "/livez" ] || ! echo "${resources}" | grep -q '"cpu":"2"'; then - echo "::warning::StatefulSet resources/probes were reconciled away; update the ArgoCD source kustomization" + echo "effective requests=${request_cpu}/${request_memory} limits=${limit_cpu}/${limit_memory} liveness=${liveness}" + if [ "${request_cpu}" != "1" ] || [ "${request_memory}" != "512Mi" ] || \ + [ "${limit_cpu}" != "2" ] || [ "${limit_memory}" != "1Gi" ] || \ + [ "${liveness}" != "/livez" ]; then + echo "::error::ArgoCD reverted StatefulSet resources/probes. Apply deploy/argocd/im-server-test/statefulset-performance-patch.yaml to the im-server-test source." + exit 1 fi diff --git a/deploy/argocd/im-server-test/README.md b/deploy/argocd/im-server-test/README.md new file mode 100644 index 0000000..47d49d8 --- /dev/null +++ b/deploy/argocd/im-server-test/README.md @@ -0,0 +1,21 @@ +# im-server-test ArgoCD patch + +The live StatefulSet is owned by the ArgoCD Application `im-server-test`. +Copy `statefulset-performance-patch.yaml` into that application's Git source +and reference it from the test overlay: + +```yaml +patches: + - path: statefulset-performance-patch.yaml +``` + +After ArgoCD sync, verify the effective configuration: + +```bash +kubectl -n im-server-system get statefulset im-server -o jsonpath='{.spec.template.spec.containers[0].resources}' +kubectl -n im-server-system get statefulset im-server -o jsonpath='{.spec.template.spec.containers[0].livenessProbe.httpGet.path}' +``` + +Expected values are requests `1 CPU / 512Mi`, limits `2 CPU / 1Gi`, and +liveness path `/livez`. The application repository's deploy workflow waits +for reconciliation and fails if ArgoCD restores different values. diff --git a/deploy/argocd/im-server-test/statefulset-performance-patch.yaml b/deploy/argocd/im-server-test/statefulset-performance-patch.yaml new file mode 100644 index 0000000..d0e5693 --- /dev/null +++ b/deploy/argocd/im-server-test/statefulset-performance-patch.yaml @@ -0,0 +1,42 @@ +apiVersion: apps/v1 +kind: StatefulSet +metadata: + name: im-server + namespace: im-server-system +spec: + template: + spec: + containers: + - name: im-server + env: + - name: CLUSTER_PROXY_TO_MASTER_BUFFER + value: "4096" + - name: CLUSTER_PROXY_TO_MASTER_WORKERS + value: "16" + - name: CLUSTER_PROXY_TO_MASTER_ENQUEUE_TIMEOUT_MS + value: "300" + - name: CLUSTER_ROUTE_RETRIES + value: "3" + - name: CLUSTER_ROUTE_RETRY_BACKOFF_MS + value: "25" + resources: + requests: + cpu: "1" + memory: 512Mi + limits: + cpu: "2" + memory: 1Gi + readinessProbe: + httpGet: + path: /healthz + port: 6060 + timeoutSeconds: 3 + periodSeconds: 10 + failureThreshold: 3 + livenessProbe: + httpGet: + path: /livez + port: 6060 + timeoutSeconds: 3 + periodSeconds: 20 + failureThreshold: 3 diff --git a/docs/IM-监控运维对接文档.md b/docs/IM-监控运维对接文档.md index 0febf61..d9a1736 100644 --- a/docs/IM-监控运维对接文档.md +++ b/docs/IM-监控运维对接文档.md @@ -123,7 +123,7 @@ HEALTHCHECK --interval=1m --timeout=3s --start-period=20s \ | 探针 | 路径 | 初始延迟 | 周期 | 超时 | 失败次数 | |---|---|---:|---:|---:|---:| | im-server readinessProbe | `/healthz` | 15s | 10s | 3s | 3 | -| im-server livenessProbe | `/healthz` | 30s | 30s | 3s | 3 | +| im-server livenessProbe | `/livez` | 15s | 20s | 3s | 3 | | im-manager readinessProbe | `/healthz` | 15s | 10s | 3s | 3 | | im-manager livenessProbe | `/healthz` | 30s | 30s | 3s | 3 | diff --git a/docs/test/IM-单聊压测Pod日志根因分析与改进方案-20260927.md b/docs/test/IM-单聊压测Pod日志根因分析与改进方案-20260927.md index be04d86..2b8e3be 100644 --- a/docs/test/IM-单聊压测Pod日志根因分析与改进方案-20260927.md +++ b/docs/test/IM-单聊压测Pod日志根因分析与改进方案-20260927.md @@ -297,3 +297,28 @@ Run ID：`p2p-200-20260927-01`，A/B 两台压测机各模拟 200 个用户， - ACK P95 和到达 P95 均落在 2 秒直方图上界，未达 500ms / 1s 门槛。 - ArgoCD 已将 StatefulSet 资源回滚为 `request 100m/128Mi`、`limit 500m/256Mi`，且 liveness 仍使用 `/healthz`。压测期间存在 CPU throttling 累计，部署配置必须在 GitOps 声明源中固化，否则应用后仍会被回滚。 - 403 修复发布后应重跑 200 对；只有 ACK 成功率、实时到达率、MySQL 精确落库均达标，才进入 500 对档位。 + +## 12. 200 对第三轮复测结果 + +Run ID：`p2p-200-20260927-03`。压测前修复了压测工具复用固定 `Idempotency-Key` 而跳过好友关系校验的问题，并重新核对、修复 200 对测试账号。 + +| 指标 | 结果 | +|---|---:| +| WS Ready | 400/400 | +| Publish 尝试 / ACK | 4000 / 4000 | +| Publish 拒绝 / 403 / 502 / 503 | 0 / 0 / 0 / 0 | +| ACK 后应到达 / 实际到达 | 4000 / 4000 | +| 实时到达率 | 100% | +| 历史查询 | 4000/4000 | +| MySQL 精确匹配 | 4000/4000 | +| MySQL 缺失 / 错配 / 重复 / 非 ACK 落库 | 0 / 0 / 0 / 0 | +| 丢失 / 重复 / 断连 | 0 / 0 / 0 | +| 乱序 | 1 | +| Cluster 队列采样最大值 | 0 | +| Kafka 峰值 lag / 最旧 backlog age / 最终 lag | 3028 / 92.624s / 0 | + +结论：跨 Pod 发送、ACK 后实时投递、Kafka 最终消费和 MySQL 落库的数据正确性已通过 200 对验证。结果报告仍为不通过，原因是 B 端记录 1 次单连接 Topic seq 回退，以及 ACK P95 和投递 P95 均落入 1-2s 区间。 + +资源采样显示，三个 Pod 峰值内存低于 94MiB，但 Pod 0/1/2 在本轮窗口内的 CPU throttled 增量约为 7.62s / 0.78s / 8.51s。实际 StatefulSet 仍是 `request 100m/128Mi`、`limit 500m/256Mi`，且 liveness 仍是 `/healthz`。因此延迟和 Kafka 追平速度尚不能作为扩容到 500 对的验收基线。 + +已增加 `deploy/argocd/im-server-test/statefulset-performance-patch.yaml`，要求 ArgoCD 声明源固化 `request 1C/512Mi`、`limit 2C/1Gi`和 `/livez` liveness。部署流水线改为等待 ArgoCD 回写后严格校验，如果再被回滚则直接失败。压测工具保留乱序为失败门禁，并新增最多 100 条 `receiver/topic_hash/previous_seq/current_seq/client_mid` 明细，用于下轮精确定位。 diff --git a/loadtest/im-pressure/merge.go b/loadtest/im-pressure/merge.go index ddab5e9..3482325 100644 --- a/loadtest/im-pressure/merge.go +++ b/loadtest/im-pressure/merge.go @@ -10,34 +10,35 @@ import ( ) type CombinedReport struct { - RunID string `json:"run_id"` - Scenario string `json:"scenario"` - GeneratedAt time.Time `json:"generated_at"` - Agents []AgentSummary `json:"agents"` - ExpectedConnections int `json:"expected_connections"` - ReadyConnections int `json:"ready_connections"` - PublishAttempted int64 `json:"publish_attempted"` - PublishAccepted int64 `json:"publish_accepted"` - PublishRejected int64 `json:"publish_rejected"` - PublishErrors map[string]int64 `json:"publish_errors,omitempty"` - PublishSuccessRate float64 `json:"publish_success_rate"` - AckP95MS float64 `json:"ack_p95_ms"` - DeliveryP95MS float64 `json:"delivery_p95_ms"` - 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"` + RunID string `json:"run_id"` + Scenario string `json:"scenario"` + GeneratedAt time.Time `json:"generated_at"` + Agents []AgentSummary `json:"agents"` + ExpectedConnections int `json:"expected_connections"` + ReadyConnections int `json:"ready_connections"` + PublishAttempted int64 `json:"publish_attempted"` + PublishAccepted int64 `json:"publish_accepted"` + PublishRejected int64 `json:"publish_rejected"` + PublishErrors map[string]int64 `json:"publish_errors,omitempty"` + PublishSuccessRate float64 `json:"publish_success_rate"` + AckP95MS float64 `json:"ack_p95_ms"` + DeliveryP95MS float64 `json:"delivery_p95_ms"` + 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"` + OrderingViolations []OrderingViolation `json:"ordering_violations,omitempty"` + 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 { @@ -73,6 +74,7 @@ func runMerge(aPath, bPath, output string) error { report.RealtimeLostStored += summary.RealtimeLostStored report.Duplicates += summary.Duplicates report.OutOfOrder += summary.OutOfOrder + report.OrderingViolations = append(report.OrderingViolations, summary.OrderingViolations...) report.Disconnects += summary.Disconnects report.HistoryChecked += summary.HistoryChecked report.HistoryFound += summary.HistoryFound diff --git a/loadtest/im-pressure/merge_test.go b/loadtest/im-pressure/merge_test.go index 33c3864..3c2d0c9 100644 --- a/loadtest/im-pressure/merge_test.go +++ b/loadtest/im-pressure/merge_test.go @@ -16,7 +16,8 @@ func TestRunMergeComputesArrivalAndPersistenceGate(t *testing.T) { b := AgentSummary{RunID: "run1", Scenario: "p2p", Machine: "b", ReadyConnections: 1000, ExpectedConnections: 1000, PublishAttempted: 10, PublishAccepted: 10, ExpectedPeer: 10, ReceivedPeer: 10, - HistoryChecked: 10, HistoryFound: 9, PersistenceLost: 1} + HistoryChecked: 10, HistoryFound: 9, PersistenceLost: 1, OutOfOrder: 1, + OrderingViolations: []OrderingViolation{{ReceiverMachine: "b", ReceiverIndex: 7, PreviousSeq: 12, CurrentSeq: 11}}} 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) @@ -37,6 +38,9 @@ func TestRunMergeComputesArrivalAndPersistenceGate(t *testing.T) { if report.Passed || report.PersistenceLost != 1 { t.Fatalf("correctness gate did not fail: %+v", report) } + if len(report.OrderingViolations) != 1 || report.OrderingViolations[0].ReceiverIndex != 7 { + t.Fatalf("ordering violations were not merged: %+v", report.OrderingViolations) + } } func TestMergedHistogramP95(t *testing.T) { diff --git a/loadtest/im-pressure/model.go b/loadtest/im-pressure/model.go index f1fbc06..5f6578b 100644 --- a/loadtest/im-pressure/model.go +++ b/loadtest/im-pressure/model.go @@ -343,39 +343,49 @@ type AcceptedMessage struct { } 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"` - ExpectedConnections int `json:"expected_connections"` - ReadyConnections int `json:"ready_connections"` - PublishAttempted int64 `json:"publish_attempted"` - PublishAccepted int64 `json:"publish_accepted"` - PublishRejected int64 `json:"publish_rejected"` - PublishErrors map[string]int64 `json:"publish_errors,omitempty"` - 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"` + 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"` + ExpectedConnections int `json:"expected_connections"` + ReadyConnections int `json:"ready_connections"` + PublishAttempted int64 `json:"publish_attempted"` + PublishAccepted int64 `json:"publish_accepted"` + PublishRejected int64 `json:"publish_rejected"` + PublishErrors map[string]int64 `json:"publish_errors,omitempty"` + 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"` + OrderingViolations []OrderingViolation `json:"ordering_violations,omitempty"` + 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 OrderingViolation struct { + ReceiverMachine string `json:"receiver_machine"` + ReceiverIndex int `json:"receiver_index"` + TopicHash string `json:"topic_hash"` + PreviousSeq int64 `json:"previous_seq"` + CurrentSeq int64 `json:"current_seq"` + ClientMid string `json:"client_mid"` } type MissingDelivery struct { diff --git a/loadtest/im-pressure/recorder.go b/loadtest/im-pressure/recorder.go index 571cf88..503a1e1 100644 --- a/loadtest/im-pressure/recorder.go +++ b/loadtest/im-pressure/recorder.go @@ -30,6 +30,7 @@ type deliveryRecorder struct { lastSeq []map[string]int64 duplicates int64 outOfOrder int64 + ordering []OrderingViolation malformed int64 history map[string]bool } @@ -92,6 +93,16 @@ func (r *deliveryRecorder) onData(account AccountRecord, data dataPacket) { r.received.Add(1) if previous := r.lastSeq[index][data.Topic]; previous > 0 && data.Seq < previous { r.outOfOrder++ + if len(r.ordering) < 100 { + r.ordering = append(r.ordering, OrderingViolation{ + ReceiverMachine: r.machine, + ReceiverIndex: account.Index, + TopicHash: topicHash(data.Topic), + PreviousSeq: previous, + CurrentSeq: data.Seq, + ClientMid: mid, + }) + } } if data.Seq > r.lastSeq[index][data.Topic] { r.lastSeq[index][data.Topic] = data.Seq @@ -132,6 +143,7 @@ func (r *deliveryRecorder) summarize(base AgentSummary, ledger []AcceptedMessage defer r.mu.Unlock() base.Duplicates = r.duplicates base.OutOfOrder = r.outOfOrder + base.OrderingViolations = append([]OrderingViolation(nil), r.ordering...) if r.malformed > 0 { base.Errors = append(base.Errors, fmt.Sprintf("malformed run messages=%d", r.malformed)) } diff --git a/loadtest/im-pressure/recorder_test.go b/loadtest/im-pressure/recorder_test.go index 6eda919..ebc1672 100644 --- a/loadtest/im-pressure/recorder_test.go +++ b/loadtest/im-pressure/recorder_test.go @@ -48,6 +48,24 @@ func TestDeliveryRecorderSeparatesHistoryFromRealtime(t *testing.T) { } } +func TestDeliveryRecorderCapturesOrderingViolationDetails(t *testing.T) { + recorder := newDeliveryRecorder("run1", "b", 1) + newer := AcceptedMessage{RunID: "run1", ClientMid: "run1-p-a-0001-02", Topic: "p2p-one", Seq: 12} + older := AcceptedMessage{RunID: "run1", ClientMid: "run1-p-a-0001-01", Topic: "p2p-one", Seq: 11} + recorder.onData(AccountRecord{Index: 1}, testData(newer, time.Now())) + recorder.onData(AccountRecord{Index: 1}, testData(older, time.Now())) + + summary := recorder.summarize(AgentSummary{}, nil) + if summary.OutOfOrder != 1 || len(summary.OrderingViolations) != 1 { + t.Fatalf("out_of_order=%d violations=%+v", summary.OutOfOrder, summary.OrderingViolations) + } + violation := summary.OrderingViolations[0] + if violation.ReceiverMachine != "b" || violation.ReceiverIndex != 1 || + violation.PreviousSeq != 12 || violation.CurrentSeq != 11 || violation.ClientMid != older.ClientMid { + t.Fatalf("unexpected violation: %+v", violation) + } +} + func testData(message AcceptedMessage, sentAt time.Time) dataPacket { content, _ := json.Marshal(map[string]any{"run_id": message.RunID, "sent_at_unix_nano": sentAt.UnixNano()}) return dataPacket{Topic: message.Topic, Seq: message.Seq,