From bde02e8eacac165efe08d1479f1340e63fd0bea0 Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Sun, 27 Sep 2026 21:08:35 +0800 Subject: [PATCH] perf: expose and reduce message publish latency --- ...压测Pod日志根因分析与改进方案-20260927.md | 31 ++++++++ loadtest/im-pressure/README.md | 4 +- loadtest/im-pressure/merge.go | 71 +++++++++++-------- loadtest/im-pressure/merge_test.go | 31 ++++++++ loadtest/im-pressure/model.go | 71 ++++++++++--------- loadtest/im-pressure/recorder.go | 65 ++++++++++++----- loadtest/im-pressure/recorder_test.go | 17 +++++ loadtest/im-pressure/wsclient.go | 3 +- server/kafka_message_persist.go | 29 ++++++-- server/main.go | 5 ++ server/meetingim.conf | 1 + server/metrics/metrics.go | 17 ++++- server/session.go | 14 ++-- server/store/store.go | 2 +- server/topic.go | 11 +++ 15 files changed, 274 insertions(+), 98 deletions(-) diff --git a/docs/test/IM-单聊压测Pod日志根因分析与改进方案-20260927.md b/docs/test/IM-单聊压测Pod日志根因分析与改进方案-20260927.md index a624926..0c61b03 100644 --- a/docs/test/IM-单聊压测Pod日志根因分析与改进方案-20260927.md +++ b/docs/test/IM-单聊压测Pod日志根因分析与改进方案-20260927.md @@ -482,3 +482,34 @@ Run ID：`p2p500-fix-20260927-02`。A/B 各建立 500 条 WebSocket，共 500 - `tenant.updated`、`tenant.status.changed`、`tenant.limits.updated`、`tenant.features.updated` 和 `cache.tenant.invalidate` 继续使用现有 Kafka 广播事件；`InvalidateTenantConfigCache` 现在同时使数据库路由缓存失效。失效与刷新并发时通过 generation 校验阻止旧查询重新写入有效缓存。 - 压测配置新增 `history_settle`，默认 5 分钟。实时 drain 完成后先等待该窗口，再发起历史查询，从而单独测量 Kafka 自然排空时间。 - 历史查询默认并发从每机 48 降为 12；历史读取高并发能力后续使用独立读写混合场景验证，不再污染消息落库吞吐结果。 + +## 17. 路由缓存发布后 500 对复测与延迟改造 + +Run ID：`p2p500-r3-20260927-03`。A/B 各 500 条 WebSocket、每人发送 10 条消息，总计 10000 次 Publish。 + +| 指标 | 结果 | +|---|---:| +| WS Ready | 1000/1000 | +| Publish ACK | 10000/10000 | +| 实时到达 / 丢失 | 10000/10000 / 0 | +| 历史回查 / 落库丢失 | 10000/10000 / 0 | +| 重复 / 乱序 / 断连 | 0 / 0 / 0 | +| 总体精确 ACK P95 | 1308.150 ms | +| A / B 精确 ACK P95 | 1121.2 / 1519.2 ms | +| 原报告投递 P95 | 2000 ms（受双机时钟偏差污染） | +| Kafka 最终 lag | 0 | +| `hotgo` pool wait | 0 | + +租户路由缓存已经消除 `hotgo` 热路径等待，消息正确性和最终落库全部通过。剩余 ACK 延迟集中在消息 4–6 的突发积压阶段。发压窗口内，`meeting_im` 连接池出现明显等待；同时 Kafka persistence consumer 正在使用同一连接池。Cluster Route RPC 平均约 1–2ms，P2M 队列始终为 0，因此跨 Pod RPC 不是秒级延迟来源。 + +原投递 P95 不能作为服务端结论：压测程序使用发送机绝对时间戳在接收机计算单向耗时。A 只接受了 5777 个非负样本，而 B 接受了全部 10000 个样本，说明 B 时钟领先 A；A 丢弃负值，B 则把时钟偏差计入延迟。 + +本次完成以下改造： + +1. 压测程序新增同机与跨机两个投递直方图；`delivery_local_p95_ms` 作为门禁，跨机负延迟计入 `cross_machine_negative_latency_samples`，不再静默丢弃后继续生成误导性 P95。 +2. 压测消息显式携带 `sender_machine`，旧消息仍可从 `client_mid` 兼容识别来源。 +3. 好友关系首次数据库确认后双向回填 Redis，减少双向聊天后续回源。 +4. 新增 `im_message_publish_stage_duration_seconds{kind,stage,result}`，覆盖 `relationship_check`、`redis_idempotency`、`redis_sequence`、`kafka_produce` 和整体 `save` 阶段。 +5. 新增 `log_client_packets` 配置，默认关闭逐包 INFO 日志，避免每条消息额外 JSON 解析和容器 stdout I/O；需要诊断时才短期开启。 + +下一轮仍使用 500 对复测。重点比较 `relationship_check`、`kafka_produce` 和 `save` 的 P95，并以 `delivery_local_p95_ms` 判断投递延迟；跨机延迟只有在 chrony/NTP 偏差确认小于 5ms 后才用于验收。数据库连接池暂不扩大，避免再次超过 MySQL 全局 170 连接预算；若 `relationship_check` 已稳定但 pool wait 与 Kafka lag 仍持续增长，再设计 persistence 独立连接预算或批量更新方案。 diff --git a/loadtest/im-pressure/README.md b/loadtest/im-pressure/README.md index 724aa47..7dd32c1 100644 --- a/loadtest/im-pressure/README.md +++ b/loadtest/im-pressure/README.md @@ -40,7 +40,7 @@ - 如果由机器 A 集中采集 agent 指标，机器 A 必须能访问 `http://机器B内网IP:9105/metrics`。 - 两台机器都必须能访问配置中的 HTTP API 和各自的 WS 地址。 - `8090` 是无鉴权的压测协调端口，只允许压测内网访问，不要暴露到公网。 -- 两台机器必须使用 NTP/chrony 同步时间；端到端延迟依赖两边时钟一致。 +- 两台机器必须使用 NTP/chrony 同步时间。报告会分别输出同机时钟延迟和跨机器单向延迟；跨机器出现负延迟样本时，后者仅用于诊断，不参与门禁。 ## 统计口径 @@ -51,6 +51,8 @@ - 每轮建立 WS 后、正式计时前，以 32 个 worker 对本机 1000 个账号做好友关系、P2P Topic 和发布权限预检；任一异常会终止本轮。 - 历史回查默认每机使用 12 个 worker、总超时 3 分钟，并在超时内持续重试缺失区间。历史读取并发能力应使用独立场景测试，不与落库吞吐测试混在一起。 - publish 错误完整分类为 `403`、`502`、`503`、`ctrl_timeout`、`socket_error` 和 `other`，不受文本错误样本上限影响。 +- `delivery_local_p95_ms` 使用发送方本机收到 echo 的同机时钟样本，是投递延迟门禁口径；兼容字段 `delivery_p95_ms` 与它保持一致。 +- `delivery_cross_machine_p95_ms` 是跨机器单向观测值，只有两机时钟同步达标时才有意义；`cross_machine_negative_latency_samples > 0` 明确表示存在时钟偏差。 - `manifest.json` 含登录 token，文件权限为 `0600`，不得提交到 Git 或放入结果包。 ## 两台机器初始化 diff --git a/loadtest/im-pressure/merge.go b/loadtest/im-pressure/merge.go index b40982c..f99f2d6 100644 --- a/loadtest/im-pressure/merge.go +++ b/loadtest/im-pressure/merge.go @@ -10,35 +10,38 @@ 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"` - 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"` + 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"` + DeliveryLocalP95MS float64 `json:"delivery_local_p95_ms"` + DeliveryCrossP95MS float64 `json:"delivery_cross_machine_p95_ms"` + CrossClockSkewSamples int64 `json:"cross_machine_negative_latency_samples"` + 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 { @@ -79,6 +82,7 @@ func runMerge(aPath, bPath, output string) error { report.HistoryChecked += summary.HistoryChecked report.HistoryFound += summary.HistoryFound report.PersistenceLost += summary.PersistenceLost + report.CrossClockSkewSamples += summary.CrossClockSkewSamples 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))) @@ -88,7 +92,14 @@ func runMerge(aPath, bPath, output string) error { report.PublishSuccessRate = float64(report.PublishAccepted) / float64(report.PublishAttempted) } report.AckP95MS = mergedAckP95(report.Agents) - report.DeliveryP95MS = mergedHistogramP95(report.Agents, func(summary AgentSummary) HistogramSnapshot { return summary.DeliveryLatency }) + report.DeliveryLocalP95MS = mergedHistogramP95(report.Agents, func(summary AgentSummary) HistogramSnapshot { return summary.DeliveryLatencyLocal }) + report.DeliveryCrossP95MS = mergedHistogramP95(report.Agents, func(summary AgentSummary) HistogramSnapshot { return summary.DeliveryLatencyCross }) + // delivery_p95_ms remains the compatibility field, but now uses only same-clock + // samples. Cross-machine wall clocks cannot produce a valid one-way latency. + report.DeliveryP95MS = report.DeliveryLocalP95MS + if report.DeliveryP95MS == 0 { + report.DeliveryP95MS = mergedHistogramP95(report.Agents, func(summary AgentSummary) HistogramSnapshot { return summary.DeliveryLatency }) + } if report.ExpectedDeliveries > 0 { report.ArrivalRate = float64(report.ReceivedDeliveries) / float64(report.ExpectedDeliveries) report.LossRate = float64(report.RealtimeLost) / float64(report.ExpectedDeliveries) diff --git a/loadtest/im-pressure/merge_test.go b/loadtest/im-pressure/merge_test.go index eb52137..2f1dd81 100644 --- a/loadtest/im-pressure/merge_test.go +++ b/loadtest/im-pressure/merge_test.go @@ -80,6 +80,37 @@ func TestMergedAckP95FallsBackWhenRawSamplesAreIncomplete(t *testing.T) { } } +func TestRunMergeUsesSameClockDeliveryLatencyForGate(t *testing.T) { + dir := t.TempDir() + local := HistogramSnapshot{Count: 10, MaxMS: 200, Buckets: map[string]int64{"200ms": 10}} + cross := HistogramSnapshot{Count: 10, MaxMS: 5000, Buckets: map[string]int64{"5s": 10}} + a := AgentSummary{RunID: "run1", Scenario: "p2p", Machine: "a", ExpectedConnections: 1, + ReadyConnections: 1, PublishAttempted: 1, PublishAccepted: 1, ExpectedPeer: 1, ReceivedPeer: 1, + DeliveryLatencyLocal: local, DeliveryLatencyCross: cross, CrossClockSkewSamples: 3} + b := a + b.Machine = "b" + 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.DeliveryP95MS != 200 || report.DeliveryLocalP95MS != 200 || report.DeliveryCrossP95MS != 5000 { + t.Fatalf("unexpected delivery percentiles: %+v", report) + } + if report.CrossClockSkewSamples != 6 || !report.Passed { + t.Fatalf("cross-machine clock skew must not fail the same-clock latency gate: %+v", report) + } +} + func writeSummaryForTest(t *testing.T, path string, summary AgentSummary) { t.Helper() raw, err := json.Marshal(summary) diff --git a/loadtest/im-pressure/model.go b/loadtest/im-pressure/model.go index 862b471..d9fa54a 100644 --- a/loadtest/im-pressure/model.go +++ b/loadtest/im-pressure/model.go @@ -350,40 +350,43 @@ 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"` - 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"` + 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"` + DeliveryLatencyLocal HistogramSnapshot `json:"delivery_latency_local"` + DeliveryLatencyCross HistogramSnapshot `json:"delivery_latency_cross_machine"` + CrossClockSkewSamples int64 `json:"cross_machine_negative_latency_samples"` + 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 { diff --git a/loadtest/im-pressure/recorder.go b/loadtest/im-pressure/recorder.go index 503a1e1..3c24140 100644 --- a/loadtest/im-pressure/recorder.go +++ b/loadtest/im-pressure/recorder.go @@ -17,22 +17,25 @@ type observation struct { } type deliveryRecorder struct { - machine string - localUsers int - latency *DurationHistogram - disconnects atomic.Int64 - received atomic.Int64 - runID atomic.Value - historyMode atomic.Bool + machine string + localUsers int + latency *DurationHistogram + localLatency *DurationHistogram + crossLatency *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 - ordering []OrderingViolation - malformed int64 - history map[string]bool + mu sync.Mutex + observed map[string]*observation + lastSeq []map[string]int64 + duplicates int64 + outOfOrder int64 + ordering []OrderingViolation + malformed int64 + crossClockSkewSamples int64 + history map[string]bool } func newDeliveryRecorder(runID, machine string, localUsers int) *deliveryRecorder { @@ -42,6 +45,7 @@ func newDeliveryRecorder(runID, machine string, localUsers int) *deliveryRecorde } recorder := &deliveryRecorder{ machine: machine, localUsers: localUsers, latency: NewDurationHistogram(), + localLatency: NewDurationHistogram(), crossLatency: NewDurationHistogram(), observed: make(map[string]*observation), lastSeq: lastSeq, history: make(map[string]bool), } recorder.runID.Store(runID) @@ -57,8 +61,9 @@ func (r *deliveryRecorder) onData(account AccountRecord, data dataPacket) { return } var content struct { - RunID string `json:"run_id"` - SentAt int64 `json:"sent_at_unix_nano"` + RunID string `json:"run_id"` + SenderMachine string `json:"sender_machine"` + SentAt int64 `json:"sent_at_unix_nano"` } if json.Unmarshal(data.Content, &content) != nil || content.RunID != runID { r.mu.Lock() @@ -108,10 +113,31 @@ func (r *deliveryRecorder) onData(account AccountRecord, data dataPacket) { r.lastSeq[index][data.Topic] = data.Seq } if content.SentAt > 0 { - r.latency.Observe(time.Since(time.Unix(0, content.SentAt))) + delay := time.Since(time.Unix(0, content.SentAt)) + r.latency.Observe(delay) + senderMachine := content.SenderMachine + if senderMachine == "" { + senderMachine = senderMachineFromClientMid(runID, mid) + } + if senderMachine == r.machine { + r.localLatency.Observe(delay) + } else if delay < 0 { + r.crossClockSkewSamples++ + } else { + r.crossLatency.Observe(delay) + } } } +func senderMachineFromClientMid(runID, mid string) string { + prefix := strings.TrimPrefix(mid, runID+"-") + parts := strings.SplitN(prefix, "-", 3) + if len(parts) < 2 { + return "" + } + return parts[1] +} + func (r *deliveryRecorder) metricCounts() (received, duplicates, outOfOrder, disconnects int64) { r.mu.Lock() duplicates = r.duplicates @@ -134,6 +160,8 @@ func (r *deliveryRecorder) onDisconnect(_ AccountRecord, _ error) { func (r *deliveryRecorder) summarize(base AgentSummary, ledger []AcceptedMessage) AgentSummary { base.DeliveryLatency = r.latency.Snapshot() + base.DeliveryLatencyLocal = r.localLatency.Snapshot() + base.DeliveryLatencyCross = r.crossLatency.Snapshot() base.Disconnects = r.disconnects.Load() runID, _ := r.runID.Load().(string) accepted := make(map[string]AcceptedMessage, len(ledger)) @@ -143,6 +171,7 @@ func (r *deliveryRecorder) summarize(base AgentSummary, ledger []AcceptedMessage defer r.mu.Unlock() base.Duplicates = r.duplicates base.OutOfOrder = r.outOfOrder + base.CrossClockSkewSamples = r.crossClockSkewSamples 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 ebc1672..6361e18 100644 --- a/loadtest/im-pressure/recorder_test.go +++ b/loadtest/im-pressure/recorder_test.go @@ -66,6 +66,23 @@ func TestDeliveryRecorderCapturesOrderingViolationDetails(t *testing.T) { } } +func TestDeliveryRecorderSeparatesLocalAndCrossMachineLatency(t *testing.T) { + recorder := newDeliveryRecorder("run1", "a", 1) + local := AcceptedMessage{RunID: "run1", ClientMid: "run1-p-a-0001-01", Topic: "p2p-one", Seq: 1} + cross := AcceptedMessage{RunID: "run1", ClientMid: "run1-p-b-0001-01", Topic: "p2p-one", Seq: 2} + recorder.onData(AccountRecord{Index: 1}, testData(local, time.Now().Add(-10*time.Millisecond))) + recorder.onData(AccountRecord{Index: 1}, testData(cross, time.Now().Add(time.Second))) + + summary := recorder.summarize(AgentSummary{}, nil) + if summary.DeliveryLatencyLocal.Count != 1 || summary.DeliveryLatencyCross.Count != 0 { + t.Fatalf("unexpected split latency counts: local=%d cross=%d", + summary.DeliveryLatencyLocal.Count, summary.DeliveryLatencyCross.Count) + } + if summary.CrossClockSkewSamples != 1 { + t.Fatalf("cross clock skew samples=%d", summary.CrossClockSkewSamples) + } +} + 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, diff --git a/loadtest/im-pressure/wsclient.go b/loadtest/im-pressure/wsclient.go index d61c744..558b7d9 100644 --- a/loadtest/im-pressure/wsclient.go +++ b/loadtest/im-pressure/wsclient.go @@ -245,7 +245,8 @@ func (c *WSClient) Publish(ctx context.Context, topic, clientMid, runID, scenari "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()}, + "sender_machine": c.account.Machine, "sender_index": c.account.Index, + "message_index": messageIndex, "sent_at_unix_nano": sentAt.UnixNano()}, }}) if err != nil { return AcceptedMessage{}, err diff --git a/server/kafka_message_persist.go b/server/kafka_message_persist.go index 59b1769..71025eb 100644 --- a/server/kafka_message_persist.go +++ b/server/kafka_message_persist.go @@ -53,7 +53,9 @@ func saveMessageMaybeKafka(ctx context.Context, tenantID types.TenantID, msg *ty msg.ClientMsgID = clientMsgID fromUid := types.ParseUid(msg.From) if clientMsgID != "" { + started := time.Now() claim, claimed, runtimeReady, claimErr := store.ClaimMessageUnique(tenantID, msg.Topic, fromUid, clientMsgID) + immetrics.ObserveMessagePublishStage(messageKind(msg.Topic), "redis_idempotency", started, claimErr) if claimErr != nil { return nil, claimErr } @@ -69,12 +71,16 @@ func saveMessageMaybeKafka(ctx context.Context, tenantID types.TenantID, msg *ty } } candidateMonth := msg.CreatedAt.UTC().Format("200601") - if seq, shardMonth, runtimeReady, seqErr := store.ReserveMessageSeq(tenantID, msg.Topic, msg.SeqId-1, candidateMonth); seqErr != nil { + seqStarted := time.Now() + seq, shardMonth, runtimeReady, seqErr := store.ReserveMessageSeq(tenantID, msg.Topic, msg.SeqId-1, candidateMonth) + immetrics.ObserveMessagePublishStage(messageKind(msg.Topic), "redis_sequence", seqStarted, seqErr) + if seqErr != nil { if clientMsgID != "" { _ = store.MarkMessageUniqueFailed(tenantID, msg.Topic, fromUid, clientMsgID) } return nil, seqErr - } else if runtimeReady { + } + if runtimeReady { msg.SeqId = seq if isGroupMessage { msg.ShardMonth = shardMonth @@ -83,7 +89,9 @@ func saveMessageMaybeKafka(ctx context.Context, tenantID types.TenantID, msg *ty if msg.ShardMonth == "" && isGroupMessage { msg.ShardMonth = candidateMonth } + kafkaStarted := time.Now() if err = enqueueKafkaMessagePersist(ctx, tenantID, msg, attachmentURLs, readBySender); err != nil { + immetrics.ObserveMessagePublishStage(messageKind(msg.Topic), "kafka_produce", kafkaStarted, err) if clientMsgID != "" { if markErr := store.MarkMessageUniquePending(tenantID, msg.Topic, fromUid, clientMsgID, msg.SeqId); markErr != nil { logs.Warn.Printf("kafka message persist: failed to mark message unique key pending topic=%s seq=%d: %v", msg.Topic, msg.SeqId, markErr) @@ -92,6 +100,7 @@ func saveMessageMaybeKafka(ctx context.Context, tenantID types.TenantID, msg *ty logs.Warn.Printf("kafka message persist: publish state unknown topic=%s seq=%d client_mid=%s err=%v", msg.Topic, msg.SeqId, clientMsgID, err) return &types.MessageSaveResult{SeqId: msg.SeqId, Pending: true, MarkedReadBySender: readBySender && !fromUid.IsZero()}, err } + immetrics.ObserveMessagePublishStage(messageKind(msg.Topic), "kafka_produce", kafkaStarted, nil) if clientMsgID != "" { if err = store.MarkMessageUniquePending(tenantID, msg.Topic, fromUid, clientMsgID, msg.SeqId); err != nil { logs.Warn.Printf("kafka message persist: failed to mark message unique key pending topic=%s seq=%d: %v", msg.Topic, msg.SeqId, err) @@ -105,11 +114,19 @@ func saveMessageMaybeKafka(ctx context.Context, tenantID types.TenantID, msg *ty return &types.MessageSaveResult{SeqId: msg.SeqId, Pending: true, MarkedReadBySender: readBySender && !fromUid.IsZero()}, nil } -func enqueueKafkaMessagePersist(ctx context.Context, tenantID types.TenantID, msg *types.Message, attachmentURLs []string, readBySender bool) error { - kind := "p2p" - if types.GetTopicCat(msg.Topic) == types.TopicCatGrp { - kind = "group" +func messageKind(topic string) string { + switch types.GetTopicCat(topic) { + case types.TopicCatGrp: + return "group" + case types.TopicCatP2P: + return "p2p" + default: + return "other" } +} + +func enqueueKafkaMessagePersist(ctx context.Context, tenantID types.TenantID, msg *types.Message, attachmentURLs []string, readBySender bool) error { + kind := messageKind(msg.Topic) payload, err := json.Marshal(kafkaMessagePersistPayload{ Message: *msg, AttachmentURLs: attachmentURLs, diff --git a/server/main.go b/server/main.go index fc49a22..e282b8a 100644 --- a/server/main.go +++ b/server/main.go @@ -182,6 +182,8 @@ var globals struct { tlsRedirectHTTP string // Maximum message size allowed from peer. maxMessageSize int64 + // Log complete client packets. Disabled by default because it is expensive on the publish hot path. + logClientPackets bool // Maximum number of group topic subscribers. maxSubscriberCount int // Maximum number of indexable tags. @@ -335,6 +337,8 @@ type configType struct { // Maximum message size allowed from client. Intended to prevent malicious client from sending // very large files inband (does not affect out of band uploads). MaxMessageSize int `json:"max_message_size"` + // Log every decoded client packet. Intended only for short diagnostic sessions. + LogClientPackets bool `json:"log_client_packets"` // Maximum number of group topic subscribers. MaxSubscriberCount int `json:"max_subscriber_count"` // Masked tags namespaces: tags immutable on User (mask), mutable on Topic only within the mask. @@ -670,6 +674,7 @@ func main() { if globals.maxMessageSize <= 0 { globals.maxMessageSize = defaultMaxMessageSize } + globals.logClientPackets = config.LogClientPackets // Maximum number of group topic subscribers globals.maxSubscriberCount = config.MaxSubscriberCount if globals.maxSubscriberCount <= 1 { diff --git a/server/meetingim.conf b/server/meetingim.conf index 350af04..b603150 100644 --- a/server/meetingim.conf +++ b/server/meetingim.conf @@ -10,6 +10,7 @@ "api_key_salt": "T713/rYYgW7g4m3vG6zGRh7+FM1t0T8j13koXScOAj4=", "max_message_size": 131072, + "log_client_packets": false, "max_subscriber_count": 128, "max_tag_count": 16, "permanent_accounts": false, diff --git a/server/metrics/metrics.go b/server/metrics/metrics.go index 29e2d60..1e2b714 100644 --- a/server/metrics/metrics.go +++ b/server/metrics/metrics.go @@ -107,6 +107,11 @@ var ( Help: "Kafka message persistence duration by message kind and result.", Buckets: []float64{0.001, 0.003, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10}, }, []string{"kind", "result"}) + MessagePublishStageDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "im_message_publish_stage_duration_seconds", + Help: "Message publish hot-path duration by message kind, stage, and result.", + Buckets: []float64{0.001, 0.003, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10}, + }, []string{"kind", "stage", "result"}) KafkaMessagePersistTotal = prometheus.NewCounterVec(prometheus.CounterOpts{ Name: "im_kafka_message_persist_total", Help: "Kafka message persistence outcomes by message kind and result.", @@ -130,7 +135,8 @@ func init() { HTTPRequestsTotal, HTTPRequestDuration, DBQueryDuration, RealtimeFanoutEventsTotal, RealtimeFanoutDeliveryDuration, ClusterP2MQueueLength, ClusterP2MQueueCapacity, ClusterP2MEnqueueTotal, ClusterRPCInflight, ClusterRPCDuration, - ClusterRouteTotal, ClusterProxyResponseTotal, MessagePersistDuration, KafkaMessagePersistTotal) + ClusterRouteTotal, ClusterProxyResponseTotal, MessagePersistDuration, MessagePublishStageDuration, + KafkaMessagePersistTotal) Registry.MustRegister(KafkaConsumerLag, KafkaBacklogOldestAge, KafkaDeadLettersTotal) } @@ -149,6 +155,15 @@ func ObserveDB(driverName, operation string, started time.Time, err error) { DBQueryDuration.WithLabelValues(driverName, operation, status).Observe(time.Since(started).Seconds()) } +func ObserveMessagePublishStage(kind, stage string, started time.Time, err error) { + status := "ok" + if err != nil { + status = "error" + } + MessagePublishStageDuration.WithLabelValues(normalizeLabel(kind), normalizeLabel(stage), status). + Observe(time.Since(started).Seconds()) +} + func IncRealtimeFanout(stage, kind, targetType, status string) { RealtimeFanoutEventsTotal.WithLabelValues(normalizeLabel(stage), normalizeLabel(kind), normalizeLabel(targetType), normalizeLabel(status)).Inc() diff --git a/server/session.go b/server/session.go index 1b1635b..f066875 100644 --- a/server/session.go +++ b/server/session.go @@ -519,13 +519,15 @@ func (s *Session) dispatchRaw(raw []byte) { s.queueOut(ErrMalformed("", "", now)) return } - toLog := redactClientMessageForLog(raw) - truncated := "" - if len(toLog) > 512 { - toLog = toLog[:512] - truncated = "<...>" + if globals.logClientPackets { + toLog := redactClientMessageForLog(raw) + truncated := "" + if len(toLog) > 512 { + toLog = toLog[:512] + truncated = "<...>" + } + logs.Info.Printf("in: '%s%s' sid='%s' uid='%s'", toLog, truncated, s.sid, s.uid) } - logs.Info.Printf("in: '%s%s' sid='%s' uid='%s'", toLog, truncated, s.sid, s.uid) s.dispatch(&msg) } diff --git a/server/store/store.go b/server/store/store.go index 7b353f5..f8d6f8e 100644 --- a/server/store/store.go +++ b/server/store/store.go @@ -3373,7 +3373,7 @@ func relationshipContext(tenantAdp adapter.TenantBusinessAdapter, tenantID types if friendship, err := tenantAdp.TenantFriendshipGet(tenantID, userID, peerID); err != nil { return nil, err } else if friendship != nil && friendship.Status == types.RelationshipFriend { - cacheFriendOneWay(tenantID, userID, peerID) + cacheFriendPair(tenantID, userID, peerID) ctx.Relationship = types.RelationshipFriend ctx.CanSend = true ctx.Remark = friendship.Remark diff --git a/server/topic.go b/server/topic.go index 0a4fb51..9b0a7fe 100644 --- a/server/topic.go +++ b/server/topic.go @@ -20,6 +20,7 @@ import ( "github.com/tinode/chat/server/auth" "github.com/tinode/chat/server/logs" + immetrics "github.com/tinode/chat/server/metrics" "github.com/tinode/chat/server/moderation" "github.com/tinode/chat/server/store" "github.com/tinode/chat/server/store/types" @@ -1362,7 +1363,9 @@ func (t *Topic) saveAndBroadcastMessage(msg *ClientComMessage, asUid types.Uid, store.Relations != nil && store.Store != nil && store.Store.IsOpen() { peer := t.p2pOtherUser(asUid) if !peer.IsZero() { + started := time.Now() canSend, _, err := store.Relations.CanSendP2P(t.tenantID, asUid, peer) + immetrics.ObserveMessagePublishStage("p2p", "relationship_check", started, err) if err != nil { logs.Warn.Printf("topic[%s]: failed to check p2p relationship: %v", t.name, err) msg.sess.queueOut(ErrUnknown(msg.Id, t.original(asUid), msg.Timestamp)) @@ -1375,6 +1378,7 @@ func (t *Topic) saveAndBroadcastMessage(msg *ClientComMessage, asUid types.Uid, } } + persistStarted := time.Now() saveResult, err := saveMessageMaybeKafka(context.Background(), t.tenantID, &types.Message{ ObjHeader: types.ObjHeader{CreatedAt: msg.Timestamp}, @@ -1385,6 +1389,13 @@ func (t *Topic) saveAndBroadcastMessage(msg *ClientComMessage, asUid types.Uid, Head: head, Content: content, }, attachments, (pud.modeGiven & pud.modeWant).IsReader()) + kind := "other" + if t.cat == types.TopicCatP2P { + kind = "p2p" + } else if t.cat == types.TopicCatGrp { + kind = "group" + } + immetrics.ObserveMessagePublishStage(kind, "save", persistStarted, err) if err != nil { logs.Warn.Printf("topic[%s]: failed to save message: %v", t.name, err) msg.sess.queueOut(ErrUnknown(msg.Id, t.original(asUid), msg.Timestamp))