From ce3250be646c4b1222653072e5e76d12a9315fbe Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Sun, 27 Sep 2026 03:37:20 +0800 Subject: [PATCH] fix: harden multi-pod delivery and load verification --- .gitea/workflows/deploy.yml | 16 ++ docker/README.md | 5 + docker/tinode/config.template | 5 + docker/tinode/entrypoint.sh | 5 + docs/IM-原生Cluster运维部署说明.md | 28 +++ ...压测Pod日志根因分析与改进方案-20260927.md | 14 ++ loadtest/im-pressure/README.md | 23 +- loadtest/im-pressure/agent.go | 175 ++++++++++++--- loadtest/im-pressure/agent_metrics.go | 5 + loadtest/im-pressure/collector.go | 41 +++- loadtest/im-pressure/collector_test.go | 31 +++ loadtest/im-pressure/config.example.json | 4 + loadtest/im-pressure/dbverify.go | 166 +++++++++++++++ loadtest/im-pressure/main.go | 25 ++- loadtest/im-pressure/merge.go | 58 ++++- loadtest/im-pressure/merge_test.go | 10 + loadtest/im-pressure/model.go | 55 +++-- loadtest/im-pressure/model_test.go | 21 +- monitoring/README.md | 2 + monitoring/im-server-alerts.yaml | 71 +++++++ server/cluster.go | 201 +++++++++++++++--- server/cluster_overload_test.go | 35 +++ server/db/adapter.go | 5 + server/db/mysql/tenant_adapter.go | 57 +++++ server/kafka_message_persist.go | 29 ++- server/kafka_runtime.go | 69 ++++++ server/main.go | 19 ++ server/meetingim.conf | 5 + server/metrics/metrics.go | 93 +++++++- server/monitoring_http.go | 67 ++++++ server/monitoring_http_test.go | 44 ++++ server/tinode.conf | 5 + server/topic_proxy.go | 45 +++- 33 files changed, 1340 insertions(+), 94 deletions(-) create mode 100644 loadtest/im-pressure/collector_test.go create mode 100644 loadtest/im-pressure/dbverify.go create mode 100644 monitoring/im-server-alerts.yaml create mode 100644 server/cluster_overload_test.go diff --git a/.gitea/workflows/deploy.yml b/.gitea/workflows/deploy.yml index 266805e..279fbcd 100644 --- a/.gitea/workflows/deploy.yml +++ b/.gitea/workflows/deploy.yml @@ -34,5 +34,21 @@ jobs: mkdir -p ~/.kube echo "${{ secrets.KUBECONFIG }}" > ~/.kube/config chmod 600 ~/.kube/config + kubectl set resources statefulset/im-server -n im-server-system -c im-server \ + --requests=cpu=1,memory=512Mi --limits=cpu=2,memory=1Gi + kubectl set env statefulset/im-server -n im-server-system -c im-server \ + CLUSTER_PROXY_TO_MASTER_BUFFER=4096 \ + CLUSTER_PROXY_TO_MASTER_WORKERS=16 \ + CLUSTER_PROXY_TO_MASTER_ENQUEUE_TIMEOUT_MS=300 \ + CLUSTER_ROUTE_RETRIES=3 \ + CLUSTER_ROUTE_RETRY_BACKOFF_MS=25 + kubectl patch statefulset/im-server -n im-server-system --type=strategic -p \ + '{"spec":{"template":{"spec":{"containers":[{"name":"im-server","readinessProbe":{"httpGet":{"path":"/healthz","port":6060},"timeoutSeconds":3,"periodSeconds":10,"failureThreshold":3},"livenessProbe":{"httpGet":{"path":"/livez","port":6060},"timeoutSeconds":3,"periodSeconds":20,"failureThreshold":3}}]}}}}' + kubectl patch service/im-server -n im-server-system --type=merge -p \ + '{"spec":{"sessionAffinity":"None"}}' + if kubectl get ingress/im-server -n im-server-system >/dev/null 2>&1; then + kubectl annotate ingress/im-server -n im-server-system --overwrite \ + nginx.ingress.kubernetes.io/load-balance=least_conn + fi kubectl rollout restart statefulset/im-server -n im-server-system kubectl rollout status statefulset/im-server -n im-server-system --timeout=300s diff --git a/docker/README.md b/docker/README.md index 162ece8..40c2e73 100644 --- a/docker/README.md +++ b/docker/README.md @@ -188,6 +188,11 @@ You can specify the following environment variables when issuing `docker run` co | `AWS_S3_ENDPOINT`[^2] | string | | An endpoint URL (hostname only or fully qualified URI) to override the default endpoint; can be of any S3-compatible service, such as `minio-api.x.io` | | `AWS_SECRET_ACCESS_KEY`[^2] | string | | AWS [Secret Access Key](https://aws.amazon.com/blogs/security/wheres-my-secret-access-key/) when using `s3` media handler. | | `CLUSTER_SELF` | string | | Node name if the server is running in a Tinode cluster. | +| `CLUSTER_PROXY_TO_MASTER_BUFFER` | number | `4096` | Total bounded proxy-to-master queue capacity for each peer. | +| `CLUSTER_PROXY_TO_MASTER_WORKERS` | number | `16` | Ordered queue shards and synchronous RPC workers for each peer. | +| `CLUSTER_PROXY_TO_MASTER_ENQUEUE_TIMEOUT_MS` | number | `300` | Maximum wait for a full proxy-to-master shard before returning 503. | +| `CLUSTER_ROUTE_RETRIES` | number | `3` | Number of retries after a remote server-message route is rejected as overloaded. | +| `CLUSTER_ROUTE_RETRY_BACKOFF_MS` | number | `25` | Initial exponential retry delay for remote server-message routes. | | `DEBUG_EMAIL_VERIFICATION_CODE`[^2] | string | | Enable dummy email verification code, e.g. `123456`. Disabled by default (empty string). | | `DEFAULT_COUNTRY_CODE`[^2] | string | `US` | 2-letter country code to assign to sessions by default when the country isn't specified by the client explicitly and it's impossible to infer it. | | `EXT_CONFIG`[^1] | string | | Path to external config file to use instead of the built-in one. If this parameter is used, most other variables are ignored[^1]. | diff --git a/docker/tinode/config.template b/docker/tinode/config.template index 9e0e6c9..53b8754 100644 --- a/docker/tinode/config.template +++ b/docker/tinode/config.template @@ -281,6 +281,11 @@ "cluster_config": { "self": "", "nodes": $CLUSTER_NODES_JSON, + "proxy_to_master_buffer": $CLUSTER_PROXY_TO_MASTER_BUFFER, + "proxy_to_master_workers": $CLUSTER_PROXY_TO_MASTER_WORKERS, + "proxy_to_master_enqueue_timeout_ms": $CLUSTER_PROXY_TO_MASTER_ENQUEUE_TIMEOUT_MS, + "route_retries": $CLUSTER_ROUTE_RETRIES, + "route_retry_backoff_ms": $CLUSTER_ROUTE_RETRY_BACKOFF_MS, "failover": { "enabled": true, "heartbeat": 100, diff --git a/docker/tinode/entrypoint.sh b/docker/tinode/entrypoint.sh index d8dca8f..ba61fc3 100644 --- a/docker/tinode/entrypoint.sh +++ b/docker/tinode/entrypoint.sh @@ -13,6 +13,11 @@ else CONFIG=working.config : "${REDIS_KEY_PREFIX:=im}" + : "${CLUSTER_PROXY_TO_MASTER_BUFFER:=4096}" + : "${CLUSTER_PROXY_TO_MASTER_WORKERS:=16}" + : "${CLUSTER_PROXY_TO_MASTER_ENQUEUE_TIMEOUT_MS:=300}" + : "${CLUSTER_ROUTE_RETRIES:=3}" + : "${CLUSTER_ROUTE_RETRY_BACKOFF_MS:=25}" if [ -z "$CLUSTER_NODES_JSON" ] ; then CLUSTER_NODES_JSON='[{"name":"tinode-0","addr":"tinode-0:12000"},{"name":"tinode-1","addr":"tinode-1:12001"},{"name":"tinode-2","addr":"tinode-2:12002"}]' fi diff --git a/docs/IM-原生Cluster运维部署说明.md b/docs/IM-原生Cluster运维部署说明.md index adfc27d..5880198 100644 --- a/docs/IM-原生Cluster运维部署说明.md +++ b/docs/IM-原生Cluster运维部署说明.md @@ -175,6 +175,7 @@ metadata: name: im-server namespace: default spec: + sessionAffinity: None selector: app: im-server ports: @@ -219,6 +220,33 @@ spec: fieldPath: metadata.name - name: EXT_CONFIG value: /etc/tinode/tinode.conf + - 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 + livenessProbe: + httpGet: + path: /livez + port: 6060 + timeoutSeconds: 3 volumeMounts: - name: tinode-config mountPath: /etc/tinode diff --git a/docs/test/IM-单聊压测Pod日志根因分析与改进方案-20260927.md b/docs/test/IM-单聊压测Pod日志根因分析与改进方案-20260927.md index 27dff88..3f1e757 100644 --- a/docs/test/IM-单聊压测Pod日志根因分析与改进方案-20260927.md +++ b/docs/test/IM-单聊压测Pod日志根因分析与改进方案-20260927.md @@ -245,3 +245,17 @@ Snowflake ID 超过有符号 32 位 `INT` 上限，批量插入立即报 1264。 - Kafka backlog 的精确条数无法通过当前调试 ServiceAccount 直接读取；死信 27 条只代表核验时已经完成失败重试的部分。 即使存在上述限制，现有证据已足以分别确认 Cluster 转发过载和 Kafka 持久化主键溢出。修复并完成死信/backlog 恢复前，本轮数据不能视为已持久化。 + +## 10. P0/P1 实施状态 + +2026-09-27 已完成代码与发布配置改造： + +- Cluster 使用按 `tenant_id + topic` 固定分片的 16 worker、每 peer 总队列 4096、入队超时 300ms；远端 `hub_full` 有 3 次指数退避重试，过载返回 503，节点断连保持 502。 +- TopicMaster 的客户端 publish 接收队列满时会通过 multiplex session 回 503，不再只记录日志后等待客户端超时；同类过载日志每秒限速，完整次数由指标保留。 +- Kafka 启用时，启动阶段校验基础私聊表及所有活动租户私聊分表的 `id` 必须为 `BIGINT NOT NULL AUTO_INCREMENT`，不满足时进程不会进入 Ready。 +- 增加 Cluster 队列/RPC/路由、WebSocket 连接、Kafka 持久化结果/lag/backlog age/死信及持久化耗时指标和 PrometheusRule 告警。 +- `/healthz` 检查全部 Cluster peer，并仅在队列超过 90% 持续 30 秒时摘除 readiness；新增 `/livez` 供 liveness 使用，避免 Cluster 短暂异常触发 Pod 重启。 +- 测试环境发布流水线应用 `request=1C/512Mi`、`limit=2C/1Gi`，设置 Cluster 参数、Service `sessionAffinity=None`，并在存在 `Ingress/im-server` 时设置 NGINX `least_conn`。 +- 压测程序增加全量 P2P 预检、48 worker 历史回查、3 分钟总超时、publish 错误分类、指标快照阶段标记、P95/ACK 成功率门禁及 MySQL 独立精确核验。 + +当前调试 kubeconfig 对 StatefulSet/Service/Ingress 没有 patch 权限，因此现存 Pod 资源尚未被本地会话直接修改；以上部署项会在新提交触发有权限的 Gitea deploy runner 后生效。生效后必须按第 8 节执行 200/500/1000 对阶梯复测，不能用单元测试替代容量验收。 diff --git a/loadtest/im-pressure/README.md b/loadtest/im-pressure/README.md index 6d33781..d1cd82c 100644 --- a/loadtest/im-pressure/README.md +++ b/loadtest/im-pressure/README.md @@ -48,6 +48,9 @@ - 实时到达只统计同时匹配本轮 `run_id` 和 `x-im-client-mid` 的 `{data}`。 - 群消息设置 `noecho=false`，每条服务端接受的群消息期望 2000 个连接各收到一次。 - 实时 drain 结束后通过 `{get what=data}` 按 ACK 的 Topic/Seq 回查历史，区分跨 Pod 实时漏投与数据库持久化缺失。 +- 每轮建立 WS 后、正式计时前，以 32 个 worker 对本机 1000 个账号做好友关系、P2P Topic 和发布权限预检；任一异常会终止本轮。 +- 历史回查默认使用 48 个 worker、总超时 3 分钟，并对缺失项再并发重试两轮。 +- publish 错误完整分类为 `403`、`502`、`503`、`ctrl_timeout`、`socket_error` 和 `other`，不受文本错误样本上限影响。 - `manifest.json` 含登录 token，文件权限为 `0600`，不得提交到 Git 或放入结果包。 ## 两台机器初始化 @@ -216,6 +219,8 @@ scp "${MACHINE_B_SSH}:${MACHINE_B_REPO}/loadtest/im-pressure/results/${RUN_ID}/a ```bash ./loadtest/im-pressure/im-pressure collect \ -duration 20m -interval 5s \ + -coordinator "http://${MACHINE_A_IP}:8090" \ + -coordinator-output "loadtest/im-pressure/results/${RUN_ID}/coordinator.json" \ -url agent-a=http://127.0.0.1:9105/metrics \ -url agent-b=http://${MACHINE_B_IP}:9105/metrics \ -url im=http://PROMETHEUS_HOST/api/v1/query?query=up \ @@ -227,7 +232,23 @@ scp "${MACHINE_B_SSH}:${MACHINE_B_REPO}/loadtest/im-pressure/results/${RUN_ID}/a -output "loadtest/im-pressure/results/${RUN_ID}/metrics.jsonl" ``` -机器 A 终端 A3 运行采集器。URL 含 `&` 时需用单引号包住整个 `name=url` 参数。建议起压前 2 分钟开始采集，结束后继续 5 分钟，以观察 Kafka lag、数据库落库追平时间及资源恢复。如果机器 A 无法访问机器 B 的 `9105`，就在机器 B 单独运行一份 `collect`，结束后把 JSONL 一并传回机器 A。 +机器 A 终端 A3 运行采集器。每条 JSONL 快照会自动标记 `phase=before|during|after`：coordinator 下发开始时间后进入 `during`，coordinator 结果文件生成后进入 `after`。URL 含 `&` 时需用单引号包住整个 `name=url` 参数。建议起压前 2 分钟开始采集，结束后继续 5 分钟，以观察 Kafka lag、数据库落库追平时间及资源恢复。如果机器 A 无法访问机器 B 的 `9105`，就在机器 B 单独运行一份 `collect`，结束后把 JSONL 一并传回机器 A。 + +## MySQL 独立落库核验 + +合并报告后在机器 A 使用只读 DSN 对 ACK ledger 做独立校验。工具会根据租户和消息时间自动查询 `im_NNN_messages` 及涉及的群消息月表，精确比较 `client_mid + topic + seq`，并统计缺失、错配、重复和非 ACK 消息： + +```bash +export MYSQL_DSN='readonly:password@tcp(mysql-host:3306)/meeting_im?parseTime=true' +./loadtest/im-pressure/im-pressure verify-db \ + -coordinator "loadtest/im-pressure/results/${RUN_ID}/coordinator.json" \ + -tenant allrich \ + -dsn-env MYSQL_DSN \ + -registry-db meeting_im \ + -output "loadtest/im-pressure/results/${RUN_ID}/db-verification.json" +``` + +DSN 只能通过环境变量传入，不要写入配置、命令脚本或结果包。命令在 missing、mismatch、duplicate 或 unexpected 任一计数非零时返回失败。 ## 运行前检查 diff --git a/loadtest/im-pressure/agent.go b/loadtest/im-pressure/agent.go index 47c96ef..a31d0d3 100644 --- a/loadtest/im-pressure/agent.go +++ b/loadtest/im-pressure/agent.go @@ -3,12 +3,14 @@ package main import ( "context" "encoding/json" + "errors" "fmt" "log" "os" "path/filepath" "regexp" "sort" + "strings" "sync" "sync/atomic" "time" @@ -17,16 +19,22 @@ import ( 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 + 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 + publish403 atomic.Int64 + publish502 atomic.Int64 + publish503 atomic.Int64 + publishTimeout atomic.Int64 + publishSocket atomic.Int64 + publishOther atomic.Int64 mu sync.Mutex accepted []AcceptedMessage @@ -37,6 +45,58 @@ func validScenario(scenario string) bool { return scenario == "p2p" || scenario == "group" || scenario == "mixed" } +func preflightP2P(ctx context.Context, cfg *Config, manifest *Manifest, local []AccountRecord, apiKey string) error { + accounts := manifestAccountMap(manifest) + api := newAPIClient(cfg, apiKey) + jobs := make(chan AccountRecord) + errCh := make(chan error, len(local)) + var wg sync.WaitGroup + workers := cfg.PreflightWorkers + if workers > len(local) { + workers = len(local) + } + for i := 0; i < workers; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for source := range jobs { + peerMachine := "a" + if source.Machine == "a" { + peerMachine = "b" + } + peer := accounts[accountKey(peerMachine, source.Index)] + if peer == nil { + errCh <- fmt.Errorf("p2p preflight %s-%04d: peer is missing", source.Machine, source.Index) + continue + } + conversation, err := api.ensureP2P(ctx, source, *peer) + if err == nil && (conversation.Relationship != "friend" || conversation.Topic != source.P2PTopic) { + err = fmt.Errorf("relationship=%s topic_match=%t", conversation.Relationship, conversation.Topic == source.P2PTopic) + } + if err != nil { + errCh <- fmt.Errorf("p2p preflight %s-%04d: %w", source.Machine, source.Index, err) + } + } + }() + } + for _, account := range local { + jobs <- account + } + close(jobs) + wg.Wait() + close(errCh) + failures := make([]string, 0, len(errCh)) + for err := range errCh { + if len(failures) < 20 { + failures = append(failures, err.Error()) + } + } + if len(failures) > 0 { + return fmt.Errorf("p2p permission preflight failed: %s", strings.Join(failures, "; ")) + } + return nil +} + 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") @@ -58,6 +118,11 @@ func runAgent(ctx context.Context, cfg *Config, manifest *Manifest, machine, coo for i := range accounts { accounts[i].Endpoint = cfg.Endpoints[machine][(accounts[i].Index-1)%len(cfg.Endpoints[machine])] } + if !cfg.SkipP2PPreflight { + if err = preflightP2P(ctx, cfg, manifest, accounts, apiKey); err != nil { + return err + } + } coord := newCoordinatorClient(coordinatorURL) placeholderRunID := "pending" run := &agentRun{cfg: cfg, machine: machine, clients: make(map[int]*WSClient), ack: NewDurationHistogram()} @@ -110,7 +175,8 @@ func runAgent(ctx context.Context, cfg *Config, manifest *Manifest, machine, coo 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, + PublishErrors: run.publishErrorCounts(), + AckLatency: run.ack.Snapshot(), AcceptedLocal: local, } run.mu.Lock() base.Errors = append(base.Errors, run.errors...) @@ -187,31 +253,29 @@ func (r *agentRun) reconcileHistory(ctx context.Context, ledger []AcceptedMessag 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 { + historyTimeout, _ := configDuration(r.cfg.HistoryTimeout) + historyCtx, cancel := context.WithTimeout(ctx, historyTimeout) + defer cancel() + r.issueHistoryQueries(historyCtx, queries) + _ = waitContext(historyCtx, 2*time.Second) + for round := 0; round < 2; round++ { missing := missingHistory(r.recorder, expected) - if len(missing) == 0 || time.Now().After(deadline) { - break - } - if waitContext(ctx, 5*time.Second) != nil { + if len(missing) == 0 || historyCtx.Err() != nil { break } + retryQueries := make([]historyQuery, 0, len(missing)) 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) + retryQueries = append(retryQueries, historyQuery{client: client, topic: message.Topic, + since: message.Seq, before: message.Seq + 1, limit: 1}) } } - _ = waitContext(ctx, 500*time.Millisecond) + r.issueHistoryQueries(historyCtx, retryQueries) + _ = waitContext(historyCtx, 2*time.Second) } missing := missingHistory(r.recorder, expected) missingIDs := make([]string, 0, len(missing)) @@ -222,6 +286,40 @@ func (r *agentRun) reconcileHistory(ctx context.Context, ledger []AcceptedMessag return len(expected), len(expected) - len(missing), missingIDs } +func (r *agentRun) issueHistoryQueries(ctx context.Context, queries []historyQuery) { + workers := r.cfg.HistoryWorkers + if workers > len(queries) { + workers = len(queries) + } + if workers == 0 { + return + } + jobs := make(chan historyQuery) + var wg sync.WaitGroup + for i := 0; i < workers; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for query := range jobs { + if err := query.client.GetData(ctx, query.topic, query.since, query.before, query.limit); err != nil && ctx.Err() == nil { + r.addError(fmt.Sprintf("history get topic=%s range=%d-%d: %v", topicHash(query.topic), query.since, query.before, err)) + } + } + }() + } + for _, query := range queries { + select { + case jobs <- query: + case <-ctx.Done(): + close(jobs) + wg.Wait() + return + } + } + close(jobs) + wg.Wait() +} + func missingHistory(recorder *deliveryRecorder, expected map[string]AcceptedMessage) []AcceptedMessage { missing := make([]AcceptedMessage, 0) for mid, message := range expected { @@ -413,6 +511,7 @@ func (r *agentRun) publish(ctx context.Context, client *WSClient, topic, mid str accepted, err := client.Publish(ctx, topic, mid, r.runID, r.scenario, messageIndex) if err != nil { r.rejected.Add(1) + r.classifyPublishError(err) r.mu.Lock() if len(r.errors) < 1000 { r.errors = append(r.errors, fmt.Sprintf("publish %s: %v", mid, err)) @@ -424,6 +523,32 @@ func (r *agentRun) publish(ctx context.Context, client *WSClient, topic, mid str return accepted, nil } +func (r *agentRun) classifyPublishError(err error) { + var ctrlErr *ctrlError + switch { + case errors.As(err, &ctrlErr) && ctrlErr.Code == 403: + r.publish403.Add(1) + case errors.As(err, &ctrlErr) && ctrlErr.Code == 502: + r.publish502.Add(1) + case errors.As(err, &ctrlErr) && ctrlErr.Code == 503: + r.publish503.Add(1) + case strings.Contains(err.Error(), "ctrl timeout") || errors.Is(err, context.DeadlineExceeded): + r.publishTimeout.Add(1) + case strings.Contains(err.Error(), "websocket") || strings.Contains(err.Error(), "broken pipe") || + strings.Contains(err.Error(), "connection reset"): + r.publishSocket.Add(1) + default: + r.publishOther.Add(1) + } +} + +func (r *agentRun) publishErrorCounts() map[string]int64 { + return map[string]int64{ + "403": r.publish403.Load(), "502": r.publish502.Load(), "503": r.publish503.Load(), + "ctrl_timeout": r.publishTimeout.Load(), "socket": r.publishSocket.Load(), "other": r.publishOther.Load(), + } +} + func (r *agentRun) addAccepted(message AcceptedMessage) { r.mu.Lock() r.accepted = append(r.accepted, message) diff --git a/loadtest/im-pressure/agent_metrics.go b/loadtest/im-pressure/agent_metrics.go index 875d337..45cafa5 100644 --- a/loadtest/im-pressure/agent_metrics.go +++ b/loadtest/im-pressure/agent_metrics.go @@ -39,6 +39,11 @@ func startAgentMetrics(address string, run *agentRun) (func(), error) { 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_publish_403_total", Help: "Publish commands rejected with 403.", ConstLabels: labels}, func() float64 { return float64(run.publish403.Load()) }), + prometheus.NewCounterFunc(prometheus.CounterOpts{Name: "im_load_agent_publish_502_total", Help: "Publish commands rejected with 502.", ConstLabels: labels}, func() float64 { return float64(run.publish502.Load()) }), + prometheus.NewCounterFunc(prometheus.CounterOpts{Name: "im_load_agent_publish_503_total", Help: "Publish commands rejected with 503.", ConstLabels: labels}, func() float64 { return float64(run.publish503.Load()) }), + prometheus.NewCounterFunc(prometheus.CounterOpts{Name: "im_load_agent_publish_ctrl_timeout_total", Help: "Publish commands with control-response timeout.", ConstLabels: labels}, func() float64 { return float64(run.publishTimeout.Load()) }), + prometheus.NewCounterFunc(prometheus.CounterOpts{Name: "im_load_agent_publish_socket_error_total", Help: "Publish commands with socket errors.", ConstLabels: labels}, func() float64 { return float64(run.publishSocket.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) diff --git a/loadtest/im-pressure/collector.go b/loadtest/im-pressure/collector.go index 78d8fb7..46d02df 100644 --- a/loadtest/im-pressure/collector.go +++ b/loadtest/im-pressure/collector.go @@ -27,6 +27,7 @@ func (n *namedURLs) Set(value string) error { type metricSnapshot struct { Timestamp time.Time `json:"timestamp"` + Phase string `json:"phase"` Name string `json:"name"` URL string `json:"url"` Status int `json:"status"` @@ -35,7 +36,8 @@ type metricSnapshot struct { Error string `json:"error,omitempty"` } -func runCollector(ctx context.Context, sources namedURLs, output string, interval, duration time.Duration) error { +func runCollector(ctx context.Context, sources namedURLs, output string, interval, duration time.Duration, + coordinatorURL, coordinatorOutput string) error { if len(sources) == 0 { return fmt.Errorf("at least one --url name=url is required") } @@ -56,8 +58,10 @@ func runCollector(ctx context.Context, sources namedURLs, output string, interva defer writer.Flush() client := &http.Client{Timeout: 10 * time.Second} deadline := time.Now().Add(duration) + phase := "before" for { - snapshots := collectOnce(ctx, client, sources) + phase = detectCollectorPhase(ctx, client, phase, coordinatorURL, coordinatorOutput) + snapshots := collectOnce(ctx, client, sources, phase) for _, snapshot := range snapshots { raw, _ := json.Marshal(snapshot) if _, err = writer.Write(append(raw, '\n')); err != nil { @@ -76,7 +80,36 @@ func runCollector(ctx context.Context, sources namedURLs, output string, interva } } -func collectOnce(ctx context.Context, client *http.Client, sources namedURLs) []metricSnapshot { +func detectCollectorPhase(ctx context.Context, client *http.Client, current, coordinatorURL, coordinatorOutput string) string { + if coordinatorOutput != "" { + if _, err := os.Stat(coordinatorOutput); err == nil { + return "after" + } + } + if coordinatorURL == "" { + return current + } + req, err := http.NewRequestWithContext(ctx, http.MethodGet, + strings.TrimRight(coordinatorURL, "/")+"/v1/status", nil) + if err != nil { + return current + } + response, err := client.Do(req) + if err != nil { + return current + } + defer response.Body.Close() + var status struct { + Start StartInfo `json:"start"` + } + if response.StatusCode != http.StatusOK || json.NewDecoder(io.LimitReader(response.Body, 1<<20)).Decode(&status) != nil || + status.Start.StartAt.IsZero() || time.Now().Before(status.Start.StartAt) { + return current + } + return "during" +} + +func collectOnce(ctx context.Context, client *http.Client, sources namedURLs, phase string) []metricSnapshot { out := make([]metricSnapshot, len(sources)) var wg sync.WaitGroup for i, source := range sources { @@ -85,7 +118,7 @@ func collectOnce(ctx context.Context, client *http.Client, sources namedURLs) [] go func() { defer wg.Done() parts := strings.SplitN(source, "=", 2) - snapshot := metricSnapshot{Timestamp: time.Now().UTC(), Name: parts[0], URL: parts[1]} + snapshot := metricSnapshot{Timestamp: time.Now().UTC(), Phase: phase, Name: parts[0], URL: parts[1]} started := time.Now() req, err := http.NewRequestWithContext(ctx, http.MethodGet, parts[1], nil) if err == nil { diff --git a/loadtest/im-pressure/collector_test.go b/loadtest/im-pressure/collector_test.go new file mode 100644 index 0000000..3622b62 --- /dev/null +++ b/loadtest/im-pressure/collector_test.go @@ -0,0 +1,31 @@ +package main + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "testing" + "time" +) + +func TestDetectCollectorPhase(t *testing.T) { + start := StartInfo{RunID: "run1", StartAt: time.Now().Add(-time.Second)} + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _ = json.NewEncoder(w).Encode(map[string]any{"start": start}) + })) + defer server.Close() + client := &http.Client{Timeout: time.Second} + if got := detectCollectorPhase(context.Background(), client, "before", server.URL, ""); got != "during" { + t.Fatalf("phase=%s", got) + } + output := filepath.Join(t.TempDir(), "coordinator.json") + if err := os.WriteFile(output, []byte("{}"), 0o600); err != nil { + t.Fatal(err) + } + if got := detectCollectorPhase(context.Background(), client, "during", server.URL, output); got != "after" { + t.Fatalf("phase=%s", got) + } +} diff --git a/loadtest/im-pressure/config.example.json b/loadtest/im-pressure/config.example.json index 4f208c5..6178d54 100644 --- a/loadtest/im-pressure/config.example.json +++ b/loadtest/im-pressure/config.example.json @@ -14,6 +14,10 @@ "prepare_workers": 20, "connect_ramp": "20s", "command_timeout": "20s", + "preflight_workers": 32, + "skip_p2p_preflight": false, + "history_workers": 48, + "history_timeout": "3m", "p2p_messages_per_user": 10, "p2p_interval": "1s", "drain_p2p": "30s", diff --git a/loadtest/im-pressure/dbverify.go b/loadtest/im-pressure/dbverify.go new file mode 100644 index 0000000..2ad056e --- /dev/null +++ b/loadtest/im-pressure/dbverify.go @@ -0,0 +1,166 @@ +package main + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" + "os" + "regexp" + "sort" + "strings" + + _ "github.com/go-sql-driver/mysql" +) + +type dbVerifyResult struct { + RunID string `json:"run_id"` + TenantCode string `json:"tenant_code"` + TenantID int64 `json:"tenant_id"` + Expected int `json:"expected"` + FoundExact int `json:"found_exact"` + Missing int `json:"missing"` + Mismatch int `json:"mismatch"` + DuplicateRows int `json:"duplicate_rows"` + Unexpected int `json:"unexpected_rows"` + Tables []string `json:"tables"` + MissingMIDs []string `json:"missing_client_mids,omitempty"` +} + +type persistedIdentity struct { + Topic string + Seq int64 +} + +var dbIdentifierRE = regexp.MustCompile(`^[A-Za-z0-9_]+$`) + +func runDBVerify(ctx context.Context, coordinatorPath, output, tenantCode, dsnEnv, registryDB string) error { + raw, err := os.ReadFile(coordinatorPath) + if err != nil { + return err + } + var coordinator struct { + Start StartInfo `json:"start"` + Ledgers map[string][]AcceptedMessage `json:"ledgers"` + } + if err = json.Unmarshal(raw, &coordinator); err != nil { + return fmt.Errorf("decode coordinator: %w", err) + } + if coordinator.Start.RunID == "" { + return fmt.Errorf("coordinator has no run id") + } + dsn := strings.TrimSpace(os.Getenv(dsnEnv)) + if dsn == "" { + return fmt.Errorf("environment variable %s is required", dsnEnv) + } + if registryDB == "" { + registryDB = strings.TrimSpace(os.Getenv("TENANT_REGISTRY_DB")) + } + if registryDB == "" { + registryDB = "meeting_im" + } + if !dbIdentifierRE.MatchString(registryDB) { + return fmt.Errorf("invalid registry database identifier") + } + db, err := sql.Open("mysql", dsn) + if err != nil { + return err + } + defer db.Close() + if err = db.PingContext(ctx); err != nil { + return err + } + var tenantID int64 + query := fmt.Sprintf("SELECT id FROM `%s`.im_tenant WHERE code=?", registryDB) + if err = db.QueryRowContext(ctx, query, tenantCode).Scan(&tenantID); err != nil { + return err + } + + expected := make(map[string]AcceptedMessage) + months := make(map[string]struct{}) + for _, ledger := range coordinator.Ledgers { + for _, message := range ledger { + if message.RunID != coordinator.Start.RunID { + continue + } + expected[message.ClientMid] = message + if message.Kind == "group" { + months[message.SentAt.UTC().Format("200601")] = struct{}{} + } + } + } + tables := []string{fmt.Sprintf("im_%03d_messages", tenantID)} + for month := range months { + tables = append(tables, fmt.Sprintf("im_%03d_group_messages_%s", tenantID, month)) + } + sort.Strings(tables) + actual := make(map[string][]persistedIdentity) + for _, table := range tables { + if !dbIdentifierRE.MatchString(table) { + return fmt.Errorf("invalid generated table name %q", table) + } + rows, queryErr := db.QueryContext(ctx, fmt.Sprintf( + "SELECT client_msg_id,topic,seqid FROM `%s` WHERE tenant_id=? AND client_msg_id LIKE ?", table), + tenantID, coordinator.Start.RunID+"-%") + if queryErr != nil { + return fmt.Errorf("query %s: %w", table, queryErr) + } + for rows.Next() { + var mid, topic string + var seq int64 + if err = rows.Scan(&mid, &topic, &seq); err != nil { + _ = rows.Close() + return err + } + actual[mid] = append(actual[mid], persistedIdentity{Topic: topic, Seq: seq}) + } + if err = rows.Close(); err != nil { + return err + } + } + result := dbVerifyResult{RunID: coordinator.Start.RunID, TenantCode: tenantCode, TenantID: tenantID, + Expected: len(expected), Tables: tables} + for mid, want := range expected { + rows := actual[mid] + if len(rows) == 0 { + result.Missing++ + result.MissingMIDs = append(result.MissingMIDs, mid) + continue + } + matched := false + for _, got := range rows { + if got.Topic == want.Topic && got.Seq == want.Seq { + matched = true + } + } + if matched { + result.FoundExact++ + } else { + result.Mismatch++ + } + if len(rows) > 1 { + result.DuplicateRows += len(rows) - 1 + } + } + for mid := range actual { + if _, ok := expected[mid]; !ok { + result.Unexpected++ + } + } + sort.Strings(result.MissingMIDs) + resultRaw, _ := json.MarshalIndent(result, "", " ") + if output == "" || output == "-" { + if _, err = fmt.Fprintln(os.Stdout, string(resultRaw)); err != nil { + return err + } + } else { + if err = writeJSONFile(output, result); err != nil { + return err + } + } + if result.Missing != 0 || result.Mismatch != 0 || result.DuplicateRows != 0 || result.Unexpected != 0 { + return fmt.Errorf("database verification failed: missing=%d mismatch=%d duplicates=%d unexpected=%d", + result.Missing, result.Mismatch, result.DuplicateRows, result.Unexpected) + } + return nil +} diff --git a/loadtest/im-pressure/main.go b/loadtest/im-pressure/main.go index 87dbfe5..db7d5d3 100644 --- a/loadtest/im-pressure/main.go +++ b/loadtest/im-pressure/main.go @@ -7,6 +7,7 @@ import ( "log" "os" "os/signal" + "strings" "syscall" "time" ) @@ -33,6 +34,8 @@ func main() { err = mergeCommand(os.Args[2:]) case "collect": err = collectCommand(ctx, os.Args[2:]) + case "verify-db": + err = verifyDBCommand(ctx, os.Args[2:]) default: usage() os.Exit(2) @@ -42,6 +45,22 @@ func main() { } } +func verifyDBCommand(ctx context.Context, args []string) error { + fs := flag.NewFlagSet("verify-db", flag.ContinueOnError) + coordinator := fs.String("coordinator", "loadtest/im-pressure/results/coordinator.json", "coordinator ledger path") + output := fs.String("output", "-", "verification JSON output path, or - for stdout") + tenant := fs.String("tenant", "", "tenant code") + dsnEnv := fs.String("dsn-env", "MYSQL_DSN", "environment variable containing the business database DSN") + registryDB := fs.String("registry-db", "", "tenant registry database; defaults to TENANT_REGISTRY_DB") + if err := fs.Parse(args); err != nil { + return err + } + if strings.TrimSpace(*tenant) == "" { + return fmt.Errorf("--tenant is required") + } + return runDBVerify(ctx, *coordinator, *output, *tenant, *dsnEnv, *registryDB) +} + 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") @@ -135,12 +154,14 @@ func collectCommand(ctx context.Context, args []string) error { 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") + coordinatorURL := fs.String("coordinator", "", "optional coordinator base URL used to label before/during phases") + coordinatorOutput := fs.String("coordinator-output", "", "coordinator JSON path used to detect the after phase") if err := fs.Parse(args); err != nil { return err } - return runCollector(ctx, urls, *output, *interval, *duration) + return runCollector(ctx, urls, *output, *interval, *duration, *coordinatorURL, *coordinatorOutput) } func usage() { - fmt.Fprintln(os.Stderr, "usage: im-pressure [flags]") + fmt.Fprintln(os.Stderr, "usage: im-pressure [flags]") } diff --git a/loadtest/im-pressure/merge.go b/loadtest/im-pressure/merge.go index 92cc683..45b1dbd 100644 --- a/loadtest/im-pressure/merge.go +++ b/loadtest/im-pressure/merge.go @@ -18,6 +18,10 @@ type CombinedReport struct { 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"` @@ -47,12 +51,16 @@ func runMerge(aPath, bPath, output string) error { 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} + report := CombinedReport{RunID: a.RunID, Scenario: a.Scenario, GeneratedAt: time.Now().UTC(), Agents: []AgentSummary{a, b}, Passed: true, + PublishErrors: make(map[string]int64)} for _, summary := range report.Agents { report.ReadyConnections += summary.ReadyConnections report.PublishAttempted += summary.PublishAttempted report.PublishAccepted += summary.PublishAccepted report.PublishRejected += summary.PublishRejected + for category, count := range summary.PublishErrors { + report.PublishErrors[category] += count + } report.ExpectedDeliveries += summary.ExpectedPeer + summary.ExpectedGroup report.ReceivedDeliveries += summary.ReceivedPeer + summary.ReceivedGroup report.RealtimeLost += summary.RealtimeLost @@ -68,6 +76,11 @@ func runMerge(aPath, bPath, output string) error { report.FailureReasons = append(report.FailureReasons, fmt.Sprintf("agent %s errors=%d", summary.Machine, len(summary.Errors))) } } + if report.PublishAttempted > 0 { + report.PublishSuccessRate = float64(report.PublishAccepted) / float64(report.PublishAttempted) + } + report.AckP95MS = mergedHistogramP95(report.Agents, func(summary AgentSummary) HistogramSnapshot { return summary.AckLatency }) + report.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) @@ -75,6 +88,17 @@ func runMerge(aPath, bPath, output string) error { if report.ReadyConnections != 2000 { report.FailureReasons = append(report.FailureReasons, fmt.Sprintf("ready connections=%d expected=2000", report.ReadyConnections)) } + if report.PublishAttempted > 0 && report.PublishSuccessRate < 0.9999 { + report.FailureReasons = append(report.FailureReasons, + fmt.Sprintf("publish success rate=%.6f expected>=0.999900", report.PublishSuccessRate)) + } + if report.AckP95MS > 500 { + report.FailureReasons = append(report.FailureReasons, fmt.Sprintf("ack p95=%.0fms expected<=500ms", report.AckP95MS)) + } + if report.DeliveryP95MS > 1000 { + report.FailureReasons = append(report.FailureReasons, + fmt.Sprintf("delivery p95=%.0fms expected<=1000ms", report.DeliveryP95MS)) + } 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") } @@ -88,6 +112,38 @@ func runMerge(aPath, bPath, output string) error { return writeJSONFile(output, report) } +func mergedHistogramP95(agents []AgentSummary, selectHistogram func(AgentSummary) HistogramSnapshot) float64 { + var count int64 + buckets := make([]int64, len(latencyBounds)+1) + maxMS := float64(0) + for _, agent := range agents { + histogram := selectHistogram(agent) + count += histogram.Count + if histogram.MaxMS > maxMS { + maxMS = histogram.MaxMS + } + for i, bound := range latencyBounds { + buckets[i] += histogram.Buckets[bound.String()] + } + buckets[len(latencyBounds)] += histogram.Buckets["+Inf"] + } + if count == 0 { + return 0 + } + target := (count*95 + 99) / 100 + var cumulative int64 + for i, bucket := range buckets { + cumulative += bucket + if cumulative >= target { + if i == len(latencyBounds) { + return maxMS + } + return float64(latencyBounds[i]) / float64(time.Millisecond) + } + } + return maxMS +} + func readAgentSummary(path string) (AgentSummary, error) { var summary AgentSummary raw, err := os.ReadFile(path) diff --git a/loadtest/im-pressure/merge_test.go b/loadtest/im-pressure/merge_test.go index 6e3e00d..4eb588b 100644 --- a/loadtest/im-pressure/merge_test.go +++ b/loadtest/im-pressure/merge_test.go @@ -37,6 +37,16 @@ func TestRunMergeComputesArrivalAndPersistenceGate(t *testing.T) { } } +func TestMergedHistogramP95(t *testing.T) { + agents := []AgentSummary{ + {AckLatency: HistogramSnapshot{Count: 10, MaxMS: 900, Buckets: map[string]int64{"200ms": 9, "1s": 1}}}, + {AckLatency: HistogramSnapshot{Count: 10, MaxMS: 300, Buckets: map[string]int64{"200ms": 10}}}, + } + if got := mergedHistogramP95(agents, func(summary AgentSummary) HistogramSnapshot { return summary.AckLatency }); got != 200 { + t.Fatalf("p95=%v", got) + } +} + 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 ff82eb8..d822511 100644 --- a/loadtest/im-pressure/model.go +++ b/loadtest/im-pressure/model.go @@ -15,22 +15,26 @@ import ( const manifestVersion = 1 type Config struct { - Tenant string `json:"tenant"` - HTTPBase string `json:"http_base"` - APIKeyEnv string `json:"api_key_env"` - PasswordEnv string `json:"password_env"` - UsersPerMachine int `json:"users_per_machine"` - Endpoints map[string][]string `json:"endpoints"` - 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"` - PhonePrefix string `json:"registration_phone_prefix"` - VerifyCodeEnv string `json:"verification_code_env"` + 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"` + PhonePrefix string `json:"registration_phone_prefix"` + VerifyCodeEnv string `json:"verification_code_env"` + HistoryWorkers int `json:"history_workers"` + HistoryTimeout string `json:"history_timeout"` + PreflightWorkers int `json:"preflight_workers"` + SkipP2PPreflight bool `json:"skip_p2p_preflight"` } func loadConfig(path string) (*Config, error) { @@ -69,6 +73,24 @@ func loadConfig(path string) (*Config, error) { if cfg.DrainGroup == "" { cfg.DrainGroup = "60s" } + if cfg.HistoryWorkers <= 0 { + cfg.HistoryWorkers = 48 + } + if cfg.HistoryWorkers > 128 { + return nil, errors.New("history_workers must not exceed 128") + } + if cfg.HistoryTimeout == "" { + cfg.HistoryTimeout = "3m" + } + if _, err = configDuration(cfg.HistoryTimeout); err != nil { + return nil, fmt.Errorf("history_timeout: %w", err) + } + if cfg.PreflightWorkers <= 0 { + cfg.PreflightWorkers = 32 + } + if cfg.PreflightWorkers > 128 { + return nil, errors.New("preflight_workers must not exceed 128") + } if strings.TrimSpace(cfg.HTTPBase) == "" { return nil, errors.New("http_base is required") } @@ -331,6 +353,7 @@ type AgentSummary struct { 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"` diff --git a/loadtest/im-pressure/model_test.go b/loadtest/im-pressure/model_test.go index 893eaef..e032d2f 100644 --- a/loadtest/im-pressure/model_test.go +++ b/loadtest/im-pressure/model_test.go @@ -1,6 +1,25 @@ package main -import "testing" +import ( + "os" + "path/filepath" + "testing" +) + +func TestLoadConfigAppliesBoundedHistoryDefaults(t *testing.T) { + path := filepath.Join(t.TempDir(), "config.json") + raw := []byte(`{"http_base":"https://example.test","endpoints":{"a":["wss://a"],"b":["wss://b"]}}`) + if err := os.WriteFile(path, raw, 0o600); err != nil { + t.Fatal(err) + } + cfg, err := loadConfig(path) + if err != nil { + t.Fatal(err) + } + if cfg.HistoryWorkers != 48 || cfg.HistoryTimeout != "3m" || cfg.PreflightWorkers != 32 { + t.Fatalf("unexpected defaults: %+v", cfg) + } +} func TestNewManifestDistributesUsersAndEndpoints(t *testing.T) { cfg := &Config{Tenant: "006", UsersPerMachine: 3, Endpoints: map[string][]string{ diff --git a/monitoring/README.md b/monitoring/README.md index cf18679..3ec83b2 100644 --- a/monitoring/README.md +++ b/monitoring/README.md @@ -5,3 +5,5 @@ This directory contains code related to monitoring Tinode server. Supported moni * [InfluxDB](https://www.influxdata.com/) See [exporter/README](./exporter/README.md) for more details. + +`im-server-alerts.yaml` contains the P0/P1 cluster routing, WebSocket balance, Kafka lag, backlog age, and dead-letter alert rules for a Prometheus Operator deployment. Apply it only after the cluster has the `PrometheusRule` CRD installed. diff --git a/monitoring/im-server-alerts.yaml b/monitoring/im-server-alerts.yaml new file mode 100644 index 0000000..2f017ea --- /dev/null +++ b/monitoring/im-server-alerts.yaml @@ -0,0 +1,71 @@ +apiVersion: monitoring.coreos.com/v1 +kind: PrometheusRule +metadata: + name: im-server-p0-p1 + namespace: im-server-system +spec: + groups: + - name: im-server-cluster + rules: + - alert: IMClusterP2MQueueHigh + expr: im_cluster_p2m_queue_length / clamp_min(im_cluster_p2m_queue_capacity, 1) > 0.70 + for: 30s + labels: + severity: warning + annotations: + summary: "IM cluster proxy queue is above 70%" + - alert: IMClusterP2MQueueCritical + expr: im_cluster_p2m_queue_length / clamp_min(im_cluster_p2m_queue_capacity, 1) > 0.90 + for: 30s + labels: + severity: critical + annotations: + summary: "IM cluster proxy queue is above 90%" + - alert: IMClusterEnqueueTimeout + expr: sum by (pod, peer) (increase(im_cluster_p2m_enqueue_total{result="timeout"}[1m])) > 0 + labels: + severity: critical + annotations: + summary: "IM cluster rejected proxy-to-master enqueue" + - alert: IMClusterRouteRejected + expr: sum by (pod, peer, result) (increase(im_cluster_route_total{result=~"hub_full|peer_rejected"}[1m])) > 0 + labels: + severity: critical + annotations: + summary: "IM cluster route was rejected" + - alert: IMClusterPeerUnavailable + expr: im_cluster_live_peers < im_cluster_expected_peers + for: 15s + labels: + severity: critical + annotations: + summary: "IM server has unavailable cluster peers" + - alert: IMWebSocketConnectionsImbalanced + expr: max(im_ws_connections_active) / clamp_min(avg(im_ws_connections_active), 1) > 1.30 + for: 2m + labels: + severity: warning + annotations: + summary: "IM WebSocket connections are imbalanced across pods" + - name: im-server-persistence + rules: + - alert: IMKafkaPersistDeadLetter + expr: sum by (kind, reason) (increase(im_kafka_dead_letters_total[5m])) > 0 + labels: + severity: critical + annotations: + summary: "IM Kafka persistence produced dead letters" + - alert: IMKafkaConsumerLag + expr: max by (pod, topic, group) (im_kafka_consumer_lag) > 1000 + for: 2m + labels: + severity: warning + annotations: + summary: "IM Kafka consumer lag is above 1000" + - alert: IMKafkaBacklogOld + expr: max by (pod, topic, group) (im_kafka_backlog_oldest_age_seconds) > 60 + for: 2m + labels: + severity: critical + annotations: + summary: "IM Kafka persistence backlog is older than 60 seconds" diff --git a/server/cluster.go b/server/cluster.go index d662eee..5611436 100644 --- a/server/cluster.go +++ b/server/cluster.go @@ -4,6 +4,7 @@ import ( "encoding/gob" "encoding/json" "errors" + "hash/fnv" "net" "net/rpc" "sort" @@ -14,6 +15,7 @@ import ( "github.com/tinode/chat/server/auth" "github.com/tinode/chat/server/concurrency" "github.com/tinode/chat/server/logs" + immetrics "github.com/tinode/chat/server/metrics" "github.com/tinode/chat/server/push" rh "github.com/tinode/chat/server/ringhash" "github.com/tinode/chat/server/store/types" @@ -25,18 +27,21 @@ const ( // Default timeout before attempting to reconnect to a node. clusterDefaultReconnectTime = 200 * time.Millisecond // Number of replicas in ringhash. - clusterHashReplicas = 20 - // Buffer size for sending requests from proxy to master. - clusterProxyToMasterBuffer = 64 - // Expand buffer size by this value for nodes over the basic 3-node setup. - clusterProxyToMasterBufferPerNode = 16 - // Timeout for attempting to enqueue a proxy-to-master request when the buffer is full. - clusterP2MTimeout = 20 * time.Millisecond + clusterHashReplicas = 20 + clusterDefaultP2MBuffer = 4096 + clusterDefaultP2MWorkers = 16 + clusterDefaultP2MTimeout = 300 * time.Millisecond + clusterDefaultRouteRetries = 3 + clusterDefaultRouteRetryBackoff = 25 * time.Millisecond // Buffer size for receiving responses from other nodes, per node. clusterRpcCompletionBuffer = 64 ) -var errClusterTopicOwnerUnavailable = errors.New("cluster topic owner unavailable") +var ( + errClusterTopicOwnerUnavailable = errors.New("cluster topic owner unavailable") + errClusterOverloaded = errors.New("cluster overloaded") + errClusterLoadExceeded = errors.New("cluster load exceeded") +) // ProxyReqType is the type of proxy requests. type ProxyReqType int @@ -64,7 +69,12 @@ type clusterConfig struct { // Name of this cluster node ThisName string `json:"self"` // Deprecated: this field is no longer used. - NumProxyEventGoRoutines int `json:"-"` + NumProxyEventGoRoutines int `json:"-"` + ProxyToMasterBuffer int `json:"proxy_to_master_buffer"` + ProxyToMasterWorkers int `json:"proxy_to_master_workers"` + ProxyToMasterEnqueueTimeoutMS int `json:"proxy_to_master_enqueue_timeout_ms"` + RouteRetries int `json:"route_retries"` + RouteRetryBackoffMS int `json:"route_retry_backoff_ms"` // Failover configuration Failover *clusterFailoverConfig } @@ -99,8 +109,13 @@ type ClusterNode struct { // Buffered, clusterRpcCompletionBuffer * number_of_nodes. rpcDone chan *rpc.Call - // Channel for sending proxy to master requests; buffered, clusterProxyToMasterBuffer. - p2mSender chan *ClusterReq + // Sharded queues preserve ordering for the same tenant and topic while allowing parallel RPCs. + p2mSenders []chan *ClusterReq + p2mEnqueueTimeout time.Duration + routeRetries int + routeRetryBackoff time.Duration + overloadLogAt atomic.Int64 + overloadSuppressed atomic.Int64 } func (n *ClusterNode) asyncRpcLoop() { @@ -109,19 +124,51 @@ func (n *ClusterNode) asyncRpcLoop() { } } -func (n *ClusterNode) p2mSenderLoop() { - for req := range n.p2mSender { +func (n *ClusterNode) p2mSenderLoop(sender <-chan *ClusterReq) { + for req := range sender { if req == nil { // Stop return } if err := n.proxyToMaster(req); err != nil { - logs.Warn.Println("p2mSenderLoop: call failed", n.name, err) + if errors.Is(err, errClusterOverloaded) { + n.logOverload("proxy_to_master", req.TenantID, req.RcptTo, 0) + } else { + logs.Warn.Println("p2mSenderLoop: call failed", n.name, err) + } } + n.updateP2MQueueMetric() } } +func (n *ClusterNode) logOverload(path string, tenantID types.TenantID, topic string, seq int64) { + now := time.Now().Unix() + previous := n.overloadLogAt.Load() + if previous == now || !n.overloadLogAt.CompareAndSwap(previous, now) { + n.overloadSuppressed.Add(1) + return + } + suppressed := n.overloadSuppressed.Swap(0) + logs.Warn.Printf("cluster overload: path=%s tenant=%d topic=%s seq=%d peer=%s suppressed=%d", + path, tenantID, topic, seq, n.name, suppressed) +} + +func (n *ClusterNode) updateP2MQueueMetric() { + depth := 0 + for _, sender := range n.p2mSenders { + depth += len(sender) + } + immetrics.ClusterP2MQueueLength.WithLabelValues(clusterPodName(), n.name).Set(float64(depth)) +} + +func clusterPodName() string { + if globals.cluster != nil && globals.cluster.thisNodeName != "" { + return globals.cluster.thisNodeName + } + return "standalone" +} + // ClusterSess is a basic info on a remote session where the message was created. type ClusterSess struct { TenantID types.TenantID @@ -318,7 +365,12 @@ func (n *ClusterNode) call(proc string, req, resp any) error { return errors.New("cluster: node '" + n.name + "' not connected") } + method := proc + started := time.Now() + immetrics.ClusterRPCInflight.WithLabelValues(clusterPodName(), n.name, method).Inc() + defer immetrics.ClusterRPCInflight.WithLabelValues(clusterPodName(), n.name, method).Dec() if err := n.endpoint.Call(proc, req, resp); err != nil { + immetrics.ClusterRPCDuration.WithLabelValues(clusterPodName(), n.name, method, "error").Observe(time.Since(started).Seconds()) logs.Warn.Println("cluster: call failed", n.name, err) n.lock.Lock() @@ -331,6 +383,7 @@ func (n *ClusterNode) call(proc string, req, resp any) error { n.lock.Unlock() return err } + immetrics.ClusterRPCDuration.WithLabelValues(clusterPodName(), n.name, method, "ok").Observe(time.Since(started).Seconds()) return nil } @@ -395,29 +448,47 @@ func (n *ClusterNode) proxyToMaster(msg *ClusterReq) error { var rejected bool err := n.call("Cluster.TopicMaster", msg, &rejected) if err == nil && rejected { - err = errors.New("cluster: topic master node out of sync") + err = errClusterOverloaded } return err } // proxyToMaster forwards request from topic proxy to topic master. func (n *ClusterNode) proxyToMasterAsync(msg *ClusterReq) error { + if len(n.p2mSenders) == 0 { + return errClusterTopicOwnerUnavailable + } + sender := n.p2mSenders[clusterReqShard(msg, len(n.p2mSenders))] select { - case n.p2mSender <- msg: + case sender <- msg: + n.updateP2MQueueMetric() + immetrics.ClusterP2MEnqueueTotal.WithLabelValues(clusterPodName(), n.name, "ok").Inc() return nil default: } // Buffer is full. Wait briefly before giving up. - timer := time.NewTimer(clusterP2MTimeout) + timer := time.NewTimer(n.p2mEnqueueTimeout) defer timer.Stop() select { - case n.p2mSender <- msg: + case sender <- msg: + n.updateP2MQueueMetric() + immetrics.ClusterP2MEnqueueTotal.WithLabelValues(clusterPodName(), n.name, "ok").Inc() return nil case <-timer.C: - return errors.New("cluster: load exceeded") + immetrics.ClusterP2MEnqueueTotal.WithLabelValues(clusterPodName(), n.name, "timeout").Inc() + return errClusterLoadExceeded } } +func clusterReqShard(msg *ClusterReq, count int) int { + if count <= 1 || msg == nil { + return 0 + } + h := fnv.New32a() + _, _ = h.Write([]byte(types.TopicKey{TenantID: msg.TenantID, Topic: msg.RcptTo}.RoutingKey())) + return int(h.Sum32() % uint32(count)) +} + // masterToProxyAsync forwards response from topic master to topic proxy // in a fire-and-forget manner. func (n *ClusterNode) masterToProxyAsync(msg *ClusterResp) error { @@ -430,8 +501,32 @@ func (n *ClusterNode) masterToProxyAsync(msg *ClusterResp) error { // route routes server message within the cluster. func (n *ClusterNode) route(msg *ClusterRoute) error { - var unused bool - return n.call("Cluster.Route", msg, &unused) + for attempt := 0; attempt <= n.routeRetries; attempt++ { + var rejected bool + err := n.call("Cluster.Route", msg, &rejected) + if err != nil { + immetrics.ClusterRouteTotal.WithLabelValues(clusterPodName(), n.name, "disconnected").Inc() + return err + } + if !rejected { + immetrics.ClusterRouteTotal.WithLabelValues(clusterPodName(), n.name, "ok").Inc() + return nil + } + immetrics.ClusterRouteTotal.WithLabelValues(clusterPodName(), n.name, "peer_rejected").Inc() + if attempt < n.routeRetries { + time.Sleep(n.routeRetryBackoff * time.Duration(1< config.ProxyToMasterBuffer { + config.ProxyToMasterWorkers = config.ProxyToMasterBuffer + } + if config.ProxyToMasterEnqueueTimeoutMS <= 0 { + config.ProxyToMasterEnqueueTimeoutMS = int(clusterDefaultP2MTimeout / time.Millisecond) + } + if config.RouteRetries < 0 { + config.RouteRetries = 0 + } else if config.RouteRetries == 0 { + config.RouteRetries = clusterDefaultRouteRetries + } + if config.RouteRetryBackoffMS <= 0 { + config.RouteRetryBackoffMS = int(clusterDefaultRouteRetryBackoff / time.Millisecond) + } thisName := *self if thisName == "" { @@ -1039,6 +1164,8 @@ func clusterInit(configString json.RawMessage, self *string) int { fingerprint: time.Now().Unix(), nodes: make(map[string]*ClusterNode), proxyEventQueue: concurrency.NewGoRoutinePool(len(config.Nodes) * 5), + p2mBuffer: config.ProxyToMasterBuffer, + p2mWorkers: config.ProxyToMasterWorkers, } var nodeNames []string @@ -1052,10 +1179,13 @@ func clusterInit(configString json.RawMessage, self *string) int { } globals.cluster.nodes[host.Name] = &ClusterNode{ - address: host.Addr, - name: host.Name, - done: make(chan bool, 1), - msess: make(map[string]struct{}), + address: host.Addr, + name: host.Name, + done: make(chan bool, 1), + msess: make(map[string]struct{}), + p2mEnqueueTimeout: time.Duration(config.ProxyToMasterEnqueueTimeoutMS) * time.Millisecond, + routeRetries: config.RouteRetries, + routeRetryBackoff: time.Duration(config.RouteRetryBackoffMS) * time.Millisecond, } } @@ -1113,17 +1243,18 @@ func (c *Cluster) start() { logs.Err.Fatal(err) } - var bufferSize = clusterProxyToMasterBuffer - if len(c.nodes) > 2 { - // Expand the buffer for larger (>3 node) clusters. - bufferSize += clusterProxyToMasterBufferPerNode * (len(c.nodes) - 2) - } for _, n := range c.nodes { go n.reconnect() n.rpcDone = make(chan *rpc.Call, len(c.nodes)*clusterRpcCompletionBuffer) - n.p2mSender = make(chan *ClusterReq, bufferSize) + workers := c.p2mWorkers + perWorker := (c.p2mBuffer + workers - 1) / workers + n.p2mSenders = make([]chan *ClusterReq, workers) go n.asyncRpcLoop() - go n.p2mSenderLoop() + for i := range n.p2mSenders { + n.p2mSenders[i] = make(chan *ClusterReq, perWorker) + go n.p2mSenderLoop(n.p2mSenders[i]) + } + immetrics.ClusterP2MQueueCapacity.WithLabelValues(clusterPodName(), n.name).Set(float64(perWorker * workers)) } if c.fo != nil { @@ -1147,7 +1278,9 @@ func (c *Cluster) shutdown() { } for _, n := range c.nodes { close(n.rpcDone) - close(n.p2mSender) + for _, sender := range n.p2mSenders { + close(sender) + } } globals.cluster.proxyEventQueue.Stop() diff --git a/server/cluster_overload_test.go b/server/cluster_overload_test.go new file mode 100644 index 0000000..b49c41a --- /dev/null +++ b/server/cluster_overload_test.go @@ -0,0 +1,35 @@ +package main + +import ( + "errors" + "net/http" + "testing" + "time" + + "github.com/tinode/chat/server/store/types" +) + +func TestClusterReqShardIsStablePerTenantTopic(t *testing.T) { + req := &ClusterReq{TenantID: types.TenantID(6), RcptTo: "p2p-topic"} + want := clusterReqShard(req, 16) + for i := 0; i < 100; i++ { + if got := clusterReqShard(req, 16); got != want { + t.Fatalf("unstable shard: got=%d want=%d", got, want) + } + } + if got := clusterReqShard(req, 1); got != 0 { + t.Fatalf("single worker shard=%d", got) + } +} + +func TestClusterRouteErrorReplySeparatesOverloadFromDisconnect(t *testing.T) { + msg := &ClientComMessage{Id: "test", Timestamp: time.Now()} + overload := clusterRouteErrorReply(msg, errClusterLoadExceeded) + if overload.Ctrl == nil || overload.Ctrl.Code != http.StatusServiceUnavailable || overload.Ctrl.Text != "cluster overloaded" { + t.Fatalf("unexpected overload reply: %+v", overload.Ctrl) + } + disconnected := clusterRouteErrorReply(msg, errors.New("disconnected")) + if disconnected.Ctrl == nil || disconnected.Ctrl.Code != http.StatusBadGateway { + t.Fatalf("unexpected disconnected reply: %+v", disconnected.Ctrl) + } +} diff --git a/server/db/adapter.go b/server/db/adapter.go index 3e9d29b..fa3c570 100644 --- a/server/db/adapter.go +++ b/server/db/adapter.go @@ -265,6 +265,11 @@ type TenantMessageBatchAdapter interface { TenantGroupMessageSaveBatch(t.TenantID, []*t.Message) error } +// TenantMessageSchemaAdapter validates storage contracts required by asynchronous persistence. +type TenantMessageSchemaAdapter interface { + TenantMessageSchemaValidate([]t.TenantID) error +} + // TenantHTTPInfrastructureAdapter is the optional durable infrastructure used // by Kafka dead-letter handling. type TenantHTTPInfrastructureAdapter interface { diff --git a/server/db/mysql/tenant_adapter.go b/server/db/mysql/tenant_adapter.go index 7c3b0a9..ab4fe07 100644 --- a/server/db/mysql/tenant_adapter.go +++ b/server/db/mysql/tenant_adapter.go @@ -8,7 +8,9 @@ import ( "crypto/sha256" "database/sql" "encoding/json" + "errors" "fmt" + "regexp" "sort" "strconv" "strings" @@ -2249,6 +2251,61 @@ func (a *adapter) TenantMessageSave(id t.TenantID, msg *t.Message) error { return err } +func (a *adapter) TenantMessageSchemaValidate(tenantIDs []t.TenantID) error { + if len(tenantIDs) == 0 { + return errors.New("mysql: no active tenants available for message schema validation") + } + validated := make(map[*adapter]map[string]struct{}) + for _, tenantID := range tenantIDs { + tenantAdp, err := a.routedTenantAdapter(tenantID) + if err != nil { + return fmt.Errorf("mysql: resolve tenant %d for message schema validation: %w", tenantID, err) + } + table := tenantMessagesTable(tenantID) + if tenantAdp.tenantScopedDB { + table = "im_messages" + } + if validated[tenantAdp] == nil { + validated[tenantAdp] = make(map[string]struct{}) + } + if _, ok := validated[tenantAdp][table]; ok { + continue + } + if err = tenantAdp.validateMessageIDColumn(table); err != nil { + return err + } + validated[tenantAdp][table] = struct{}{} + } + if !a.tenantScopedDB { + if err := a.validateMessageIDColumn("im_messages"); err != nil { + return err + } + } + return nil +} + +func (a *adapter) validateMessageIDColumn(table string) error { + if !regexp.MustCompile(`^im_(?:[0-9]+_)?messages$`).MatchString(table) { + return fmt.Errorf("mysql: unsafe message table name %q", table) + } + var dataType, nullable, extra string + err := a.db.QueryRow(`SELECT DATA_TYPE,IS_NULLABLE,EXTRA FROM information_schema.COLUMNS + WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=? AND COLUMN_NAME='id'`, table). + Scan(&dataType, &nullable, &extra) + if err == sql.ErrNoRows { + return fmt.Errorf("mysql: required message table %s is missing", table) + } + if err != nil { + return fmt.Errorf("mysql: inspect %s.id: %w", table, err) + } + if !strings.EqualFold(dataType, "bigint") || !strings.EqualFold(nullable, "NO") || + !strings.Contains(strings.ToLower(extra), "auto_increment") { + return fmt.Errorf("mysql: incompatible %s.id schema: type=%s nullable=%s extra=%s; want BIGINT NOT NULL AUTO_INCREMENT", + table, dataType, nullable, extra) + } + return nil +} + func (a *adapter) TenantMessageSaveBatch(id t.TenantID, messages []*t.Message) error { tenantAdp, err := a.routedTenantAdapter(id) if err != nil { diff --git a/server/kafka_message_persist.go b/server/kafka_message_persist.go index 9259bc1..59b1769 100644 --- a/server/kafka_message_persist.go +++ b/server/kafka_message_persist.go @@ -10,6 +10,7 @@ import ( "github.com/segmentio/kafka-go" "github.com/tinode/chat/server/logs" + immetrics "github.com/tinode/chat/server/metrics" "github.com/tinode/chat/server/store" "github.com/tinode/chat/server/store/types" ) @@ -160,11 +161,18 @@ func handleKafkaMessagePersistBatch(ctx context.Context, messages []kafka.Messag groups[key] = append(groups[key], item) } for _, items := range groups { + kind := items[0].payload.MessageKind + if kind == "" { + kind = "p2p" + } + started := time.Now() batch := make([]store.PreparedMessageBatchItem, 0, len(items)) for _, item := range items { batch = append(batch, store.PreparedMessageBatchItem{Message: &item.payload.Message, ReadBySender: item.payload.ReadBySender}) } if err := store.PersistPreparedMessageBatch(items[0].tenantID, batch); err == nil { + immetrics.KafkaMessagePersistTotal.WithLabelValues(kind, "batch_ok").Add(float64(len(items))) + immetrics.MessagePersistDuration.WithLabelValues(kind, "batch_ok").Observe(time.Since(started).Seconds()) for _, item := range items { if err = finishKafkaMessagePersist(ctx, item); err != nil { if err = persistKafkaMessageIndividually(ctx, item); err != nil { @@ -174,6 +182,8 @@ func handleKafkaMessagePersistBatch(ctx context.Context, messages []kafka.Messag } continue } else { + immetrics.KafkaMessagePersistTotal.WithLabelValues(kind, "batch_failed").Add(float64(len(items))) + immetrics.MessagePersistDuration.WithLabelValues(kind, "batch_failed").Observe(time.Since(started).Seconds()) logs.Warn.Printf("kafka message batch persist failed, degrading to singles: tenant=%d kind=%s shard=%s size=%d err=%v", items[0].tenantID, items[0].payload.MessageKind, items[0].payload.Message.ShardMonth, len(items), err) } @@ -227,6 +237,11 @@ func prepareKafkaMessagePersistItem(ctx context.Context, message kafka.Message, } func persistKafkaMessageIndividually(ctx context.Context, item *kafkaMessagePersistItem) error { + kind := item.payload.MessageKind + if kind == "" { + kind = "p2p" + } + started := time.Now() operation := func() error { err := store.PersistPreparedMessageBatch(item.tenantID, []store.PreparedMessageBatchItem{{ Message: &item.payload.Message, ReadBySender: item.payload.ReadBySender, @@ -237,14 +252,24 @@ func persistKafkaMessageIndividually(ctx context.Context, item *kafkaMessagePers if err != nil && !errors.Is(err, types.ErrDuplicate) { return err } - return finishKafkaMessagePersist(ctx, item) + if err = finishKafkaMessagePersist(ctx, item); err != nil { + return err + } + immetrics.KafkaMessagePersistTotal.WithLabelValues(kind, "single_ok").Inc() + immetrics.MessagePersistDuration.WithLabelValues(kind, "single_ok").Observe(time.Since(started).Seconds()) + return nil } markFailed := func(err error) (*types.KafkaConsumeRecord, error) { + immetrics.KafkaMessagePersistTotal.WithLabelValues(kind, "retry").Inc() return store.HTTPInfrastructure.MarkKafkaConsumeFailed(item.tenantID, item.envelope.EventID, kafkaBus.config.ConsumerGroup, kafkaMessagePersistFailureReason(err), err.Error(), time.Now().UTC().Round(time.Millisecond)) } markDead := func(failed *types.KafkaConsumeRecord, err error) error { - return markKafkaDeadLetter(ctx, item.tenantID, item.envelope, failed, kafkaMessagePersistFailureReason(err), err.Error()) + immetrics.KafkaMessagePersistTotal.WithLabelValues(kind, "dead").Inc() + immetrics.MessagePersistDuration.WithLabelValues(kind, "dead").Observe(time.Since(started).Seconds()) + reason := kafkaMessagePersistFailureReason(err) + immetrics.KafkaDeadLettersTotal.WithLabelValues(kind, reason).Inc() + return markKafkaDeadLetter(ctx, item.tenantID, item.envelope, failed, reason, err.Error()) } return runKafkaMessageSingleRetries(ctx, kafkaBus.config.MessageRetries, operation, markFailed, markDead, waitKafkaRetry) } diff --git a/server/kafka_runtime.go b/server/kafka_runtime.go index 769e9d1..1070725 100644 --- a/server/kafka_runtime.go +++ b/server/kafka_runtime.go @@ -8,12 +8,14 @@ import ( "regexp" "strings" "sync" + "sync/atomic" "time" "github.com/segmentio/kafka-go" "github.com/segmentio/kafka-go/sasl" "github.com/segmentio/kafka-go/sasl/scram" "github.com/tinode/chat/server/logs" + immetrics "github.com/tinode/chat/server/metrics" "github.com/tinode/chat/server/store/types" ) @@ -155,6 +157,11 @@ func initKafkaRuntime(raw json.RawMessage) error { return nil } +func kafkaConfigured(raw json.RawMessage) bool { + var cfg kafkaConfig + return len(raw) > 0 && json.Unmarshal(raw, &cfg) == nil && cfg.Enabled +} + func normalizeKafkaConfig(cfg *kafkaConfig) { if cfg.ClientID == "" { cfg.ClientID = "im-server" @@ -333,6 +340,7 @@ func startKafkaConsumerWithGroup(topicName, groupID string, handler kafkaMessage MaxWait: time.Second, Dialer: kafkaBus.dialer(), }) + readerMetrics := startKafkaReaderMetrics(kafkaBus.ctx, reader, topic, groupID) kafkaBus.wg.Add(1) go func() { defer kafkaBus.wg.Done() @@ -346,6 +354,7 @@ func startKafkaConsumerWithGroup(topicName, groupID string, handler kafkaMessage logs.Err.Println("Kafka fetch failed:", err) continue } + readerMetrics.observe(msg.Time) commit := func(ctx context.Context, message kafka.Message) error { return reader.CommitMessages(ctx, message) } @@ -355,6 +364,7 @@ func startKafkaConsumerWithGroup(topicName, groupID string, handler kafkaMessage } logs.Err.Printf("Kafka message stopped without commit: topic=%s partition=%d offset=%d err=%v", msg.Topic, msg.Partition, msg.Offset, err) } + readerMetrics.clear() } }() logs.Info.Printf("Kafka consumer started: topic=%s group=%s", topic, groupID) @@ -373,6 +383,7 @@ func startKafkaBatchConsumer(topicName string, handler kafkaBatchHandler) { Brokers: kafkaBus.config.Brokers, GroupID: groupID, Topic: topic, StartOffset: kafka.FirstOffset, MinBytes: 1, MaxBytes: 10e6, MaxWait: time.Second, Dialer: kafkaBus.dialer(), }) + readerMetrics := startKafkaReaderMetrics(kafkaBus.ctx, reader, topic, groupID) kafkaBus.wg.Add(1) go func() { defer kafkaBus.wg.Done() @@ -387,17 +398,75 @@ func startKafkaBatchConsumer(topicName string, handler kafkaBatchHandler) { logs.Err.Println("Kafka batch fetch failed:", err) continue } + if len(messages) > 0 { + readerMetrics.observe(messages[0].Time) + } if err = consumeKafkaMessageBatch(kafkaBus.ctx, messages, handler, reader.CommitMessages, waitKafkaRetry); err != nil { if errors.Is(err, context.Canceled) || errors.Is(kafkaBus.ctx.Err(), context.Canceled) { return } logs.Err.Printf("Kafka batch stopped without commit: topic=%s size=%d err=%v", topic, len(messages), err) } + readerMetrics.clear() } }() logs.Info.Printf("Kafka batch consumer started: topic=%s group=%s size=%d wait_ms=%d", topic, groupID, kafkaBus.config.MessageBatchSize, kafkaBus.config.MessageBatchWait) } +type kafkaReaderMetrics struct { + reader *kafka.Reader + topic string + group string + oldestNano atomic.Int64 +} + +func startKafkaReaderMetrics(ctx context.Context, reader *kafka.Reader, topic, group string) *kafkaReaderMetrics { + metrics := &kafkaReaderMetrics{reader: reader, topic: topic, group: group} + kafkaBus.wg.Add(1) + go func() { + defer kafkaBus.wg.Done() + ticker := time.NewTicker(5 * time.Second) + defer ticker.Stop() + metrics.update() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + metrics.update() + } + } + }() + return metrics +} + +func (m *kafkaReaderMetrics) observe(messageTime time.Time) { + if !messageTime.IsZero() { + m.oldestNano.Store(messageTime.UnixNano()) + } + m.update() +} + +func (m *kafkaReaderMetrics) clear() { + m.oldestNano.Store(0) + m.update() +} + +func (m *kafkaReaderMetrics) update() { + stats := m.reader.Stats() + if stats.Lag >= 0 { + immetrics.KafkaConsumerLag.WithLabelValues(clusterPodName(), m.topic, m.group).Set(float64(stats.Lag)) + } + age := float64(0) + if oldest := m.oldestNano.Load(); oldest > 0 { + age = time.Since(time.Unix(0, oldest)).Seconds() + if age < 0 { + age = 0 + } + } + immetrics.KafkaBacklogOldestAge.WithLabelValues(clusterPodName(), m.topic, m.group).Set(age) +} + func fetchKafkaMessageBatch(ctx context.Context, reader *kafka.Reader, maxSize int, maxWait time.Duration) ([]kafka.Message, error) { first, err := reader.FetchMessage(ctx) if err != nil { diff --git a/server/main.go b/server/main.go index 8d961d8..fc49a22 100644 --- a/server/main.go +++ b/server/main.go @@ -34,6 +34,7 @@ import ( _ "github.com/tinode/chat/server/auth/code" _ "github.com/tinode/chat/server/auth/rest" _ "github.com/tinode/chat/server/auth/token" + adapter "github.com/tinode/chat/server/db" "github.com/tinode/chat/server/store/types" // Database backends @@ -516,6 +517,23 @@ func main() { logs.Info.Println("All done, good bye") }() statsRegisterDbStats() + if kafkaConfigured(config.Kafka) { + tenants, tenantErr := store.Tenants.ListActive(10000) + if tenantErr != nil { + logs.Err.Fatal("Failed to list tenants for Kafka message schema validation:", tenantErr) + } + tenantIDs := make([]types.TenantID, 0, len(tenants)) + for i := range tenants { + tenantIDs = append(tenantIDs, tenants[i].ID) + } + validator, ok := store.Store.GetAdapter().(adapter.TenantMessageSchemaAdapter) + if !ok { + logs.Err.Fatal("Kafka message persistence requires a message schema validator") + } + if err = validator.TenantMessageSchemaValidate(tenantIDs); err != nil { + logs.Err.Fatal("Kafka message persistence schema validation failed:", err) + } + } if err = initKafkaRuntime(config.Kafka); err != nil { logs.Err.Fatal("Failed to initialize Kafka:", err) @@ -860,6 +878,7 @@ func main() { mux.HandleFunc(sspath, serveStatus) } mux.HandleFunc("/healthz", serveHealthz) + mux.HandleFunc("/livez", serveLivez) mux.HandleFunc("/metrics", serveMetrics) // Public legal pages are opened directly by clients before login. diff --git a/server/meetingim.conf b/server/meetingim.conf index daa0ed3..9ba1926 100644 --- a/server/meetingim.conf +++ b/server/meetingim.conf @@ -238,6 +238,11 @@ "cluster_config": { "self": "", "nodes": [], + "proxy_to_master_buffer": 4096, + "proxy_to_master_workers": 16, + "proxy_to_master_enqueue_timeout_ms": 300, + "route_retries": 3, + "route_retry_backoff_ms": 25, "failover": { "enabled": false } diff --git a/server/metrics/metrics.go b/server/metrics/metrics.go index 70213e5..29e2d60 100644 --- a/server/metrics/metrics.go +++ b/server/metrics/metrics.go @@ -2,7 +2,9 @@ package metrics import ( "expvar" + "os" "strconv" + "strings" "time" "github.com/prometheus/client_golang/prometheus" @@ -45,11 +47,98 @@ var ( }, func() float64 { return float64(expvarIntValue("LiveSessions")) }) + IMWSConnectionsActive = prometheus.NewGaugeFunc(prometheus.GaugeOpts{ + Name: "im_ws_connections_active", + Help: "Current active WebSocket connections, using the IM metric namespace.", + ConstLabels: prometheus.Labels{"pod": metricPodName()}, + }, func() float64 { + return float64(expvarIntValue("LiveSessions")) + }) + IMClusterLivePeers = prometheus.NewGaugeFunc(prometheus.GaugeOpts{ + Name: "im_cluster_live_peers", + Help: "Current connected cluster peers.", + ConstLabels: prometheus.Labels{"pod": metricPodName()}, + }, func() float64 { + return float64(expvarIntValue("LiveClusterNodes")) + }) + IMClusterExpectedPeers = prometheus.NewGaugeFunc(prometheus.GaugeOpts{ + Name: "im_cluster_expected_peers", + Help: "Expected cluster peers derived from configured total nodes.", + ConstLabels: prometheus.Labels{"pod": metricPodName()}, + }, func() float64 { + total := expvarIntValue("TotalClusterNodes") + if total > 0 { + total-- + } + return float64(total) + }) + + ClusterP2MQueueLength = prometheus.NewGaugeVec(prometheus.GaugeOpts{ + Name: "im_cluster_p2m_queue_length", + Help: "Current proxy-to-master queue depth by pod and peer.", + }, []string{"pod", "peer"}) + ClusterP2MQueueCapacity = prometheus.NewGaugeVec(prometheus.GaugeOpts{ + Name: "im_cluster_p2m_queue_capacity", + Help: "Configured proxy-to-master queue capacity by pod and peer.", + }, []string{"pod", "peer"}) + ClusterP2MEnqueueTotal = prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "im_cluster_p2m_enqueue_total", + Help: "Proxy-to-master enqueue attempts by pod, peer, and result.", + }, []string{"pod", "peer", "result"}) + ClusterRPCInflight = prometheus.NewGaugeVec(prometheus.GaugeOpts{ + Name: "im_cluster_rpc_inflight", + Help: "Current synchronous cluster RPC calls by pod, peer, and method.", + }, []string{"pod", "peer", "method"}) + ClusterRPCDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "im_cluster_rpc_duration_seconds", + Help: "Cluster RPC duration by pod, peer, method, 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}, + }, []string{"pod", "peer", "method", "result"}) + ClusterRouteTotal = prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "im_cluster_route_total", + Help: "Cluster route attempts by pod, peer, and result.", + }, []string{"pod", "peer", "result"}) + ClusterProxyResponseTotal = prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "im_cluster_proxy_response_total", + Help: "Cluster proxy responses by pod and result.", + }, []string{"pod", "result"}) + MessagePersistDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "im_message_persist_duration_seconds", + 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"}) + KafkaMessagePersistTotal = prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "im_kafka_message_persist_total", + Help: "Kafka message persistence outcomes by message kind and result.", + }, []string{"kind", "result"}) + KafkaConsumerLag = prometheus.NewGaugeVec(prometheus.GaugeOpts{ + Name: "im_kafka_consumer_lag", + Help: "Kafka reader lag reported by kafka-go by pod, topic, and consumer group.", + }, []string{"pod", "topic", "group"}) + KafkaBacklogOldestAge = prometheus.NewGaugeVec(prometheus.GaugeOpts{ + Name: "im_kafka_backlog_oldest_age_seconds", + Help: "Age of the latest fetched Kafka message by pod, topic, and consumer group.", + }, []string{"pod", "topic", "group"}) + KafkaDeadLettersTotal = prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "im_kafka_dead_letters_total", + Help: "Kafka messages moved to dead letter by message kind and reason.", + }, []string{"kind", "reason"}) ) func init() { - Registry.MustRegister(WSConnectionsActive, HTTPRequestsTotal, HTTPRequestDuration, DBQueryDuration, - RealtimeFanoutEventsTotal, RealtimeFanoutDeliveryDuration) + Registry.MustRegister(WSConnectionsActive, IMWSConnectionsActive, IMClusterLivePeers, IMClusterExpectedPeers, + HTTPRequestsTotal, HTTPRequestDuration, DBQueryDuration, + RealtimeFanoutEventsTotal, RealtimeFanoutDeliveryDuration, ClusterP2MQueueLength, + ClusterP2MQueueCapacity, ClusterP2MEnqueueTotal, ClusterRPCInflight, ClusterRPCDuration, + ClusterRouteTotal, ClusterProxyResponseTotal, MessagePersistDuration, KafkaMessagePersistTotal) + Registry.MustRegister(KafkaConsumerLag, KafkaBacklogOldestAge, KafkaDeadLettersTotal) +} + +func metricPodName() string { + if value := strings.TrimSpace(os.Getenv("HOSTNAME")); value != "" { + return value + } + return "standalone" } func ObserveDB(driverName, operation string, started time.Time, err error) { diff --git a/server/monitoring_http.go b/server/monitoring_http.go index 35fad60..2f5c3b9 100644 --- a/server/monitoring_http.go +++ b/server/monitoring_http.go @@ -4,12 +4,14 @@ import ( "bufio" "context" "encoding/json" + "fmt" "io" "net" "net/http" "regexp" "strconv" "strings" + "sync" "time" "github.com/prometheus/client_golang/prometheus/promhttp" @@ -20,6 +22,8 @@ import ( var ( metricsHandler = promhttp.HandlerFor(immetrics.Registry, promhttp.HandlerOpts{}) uuidLikeMetricSegment = regexp.MustCompile(`^[0-9a-fA-F]{8}-[0-9a-fA-F-]{13,}$`) + clusterQueueHighSince = make(map[string]time.Time) + clusterQueueHealthMu sync.Mutex ) type healthzResponse struct { @@ -45,6 +49,28 @@ func serveHealthz(wrt http.ResponseWriter, req *http.Request) { } else { checks["mysql"] = "ok" } + if globals.cluster != nil { + expected := len(globals.cluster.nodes) + live := 0 + for _, node := range globals.cluster.nodes { + node.lock.Lock() + connected := node.connected + node.lock.Unlock() + if connected { + live++ + } + } + checks["cluster"] = fmt.Sprintf("%d/%d peers connected", live, expected) + if live != expected { + status = http.StatusServiceUnavailable + } + if healthy, detail := clusterQueueHealthy(globals.cluster, time.Now()); !healthy { + checks["cluster_queue"] = detail + status = http.StatusServiceUnavailable + } else { + checks["cluster_queue"] = detail + } + } body := healthzResponse{Status: "ok", Checks: checks} if status != http.StatusOK { @@ -56,6 +82,47 @@ func serveHealthz(wrt http.ResponseWriter, req *http.Request) { } } +func clusterQueueHealthy(cluster *Cluster, now time.Time) (bool, string) { + clusterQueueHealthMu.Lock() + defer clusterQueueHealthMu.Unlock() + healthy := true + detail := "ok" + for name, node := range cluster.nodes { + depth, capacity := 0, 0 + for _, sender := range node.p2mSenders { + depth += len(sender) + capacity += cap(sender) + } + if capacity == 0 || depth*100 < capacity*90 { + delete(clusterQueueHighSince, name) + continue + } + since, exists := clusterQueueHighSince[name] + if !exists { + clusterQueueHighSince[name] = now + continue + } + if now.Sub(since) >= 30*time.Second { + healthy = false + detail = fmt.Sprintf("peer %s queue %d/%d above 90%% for %s", name, depth, capacity, + now.Sub(since).Round(time.Second)) + } + } + return healthy, detail +} + +func serveLivez(wrt http.ResponseWriter, req *http.Request) { + if req.Method != http.MethodGet && req.Method != http.MethodHead { + wrt.Header().Set("Allow", http.MethodGet+", "+http.MethodHead) + http.Error(wrt, "method not allowed", http.StatusMethodNotAllowed) + return + } + wrt.WriteHeader(http.StatusOK) + if req.Method != http.MethodHead { + _, _ = io.WriteString(wrt, "ok\n") + } +} + func serveMetrics(wrt http.ResponseWriter, req *http.Request) { if req.Method != http.MethodGet && req.Method != http.MethodHead { wrt.Header().Set("Allow", http.MethodGet+", "+http.MethodHead) diff --git a/server/monitoring_http_test.go b/server/monitoring_http_test.go index c545e95..1fb5bd1 100644 --- a/server/monitoring_http_test.go +++ b/server/monitoring_http_test.go @@ -52,7 +52,44 @@ func TestServeHealthzError(t *testing.T) { } } +func TestServeLivez(t *testing.T) { + wrt := httptest.NewRecorder() + serveLivez(wrt, httptest.NewRequest(http.MethodGet, "/livez", nil)) + if wrt.Code != http.StatusOK || wrt.Body.String() != "ok\n" { + t.Fatalf("unexpected livez response: code=%d body=%q", wrt.Code, wrt.Body.String()) + } +} + +func TestClusterQueueHealthRequiresSustainedHighWatermark(t *testing.T) { + clusterQueueHealthMu.Lock() + clusterQueueHighSince = make(map[string]time.Time) + clusterQueueHealthMu.Unlock() + queue := make(chan *ClusterReq, 10) + for i := 0; i < 9; i++ { + queue <- &ClusterReq{} + } + cluster := &Cluster{nodes: map[string]*ClusterNode{"peer": {p2mSenders: []chan *ClusterReq{queue}}}} + now := time.Now() + if healthy, _ := clusterQueueHealthy(cluster, now); !healthy { + t.Fatal("instant high watermark must not fail readiness") + } + if healthy, _ := clusterQueueHealthy(cluster, now.Add(31*time.Second)); healthy { + t.Fatal("sustained high watermark must fail readiness") + } + <-queue + if healthy, _ := clusterQueueHealthy(cluster, now.Add(32*time.Second)); !healthy { + t.Fatal("queue recovery must restore readiness") + } +} + func TestServeMetricsIncludesWSConnections(t *testing.T) { + immetrics.ClusterP2MQueueLength.WithLabelValues("test", "peer").Set(0) + immetrics.ClusterP2MEnqueueTotal.WithLabelValues("test", "peer", "ok").Add(0) + immetrics.ClusterRPCDuration.WithLabelValues("test", "peer", "Cluster.Route", "ok").Observe(0) + immetrics.ClusterRouteTotal.WithLabelValues("test", "peer", "ok").Add(0) + immetrics.KafkaMessagePersistTotal.WithLabelValues("p2p", "batch_ok").Add(0) + immetrics.KafkaConsumerLag.WithLabelValues("test", "topic", "group").Set(0) + immetrics.KafkaDeadLettersTotal.WithLabelValues("p2p", "test").Add(0) wrt := httptest.NewRecorder() serveMetrics(wrt, httptest.NewRequest(http.MethodGet, "/metrics", nil)) @@ -63,6 +100,13 @@ func TestServeMetricsIncludesWSConnections(t *testing.T) { if !strings.Contains(body, "# TYPE ws_connections_active gauge") || !strings.Contains(body, "ws_connections_active ") { t.Fatalf("unexpected body %s", body) } + for _, metric := range []string{"im_ws_connections_active", "im_cluster_p2m_queue_length", + "im_cluster_p2m_enqueue_total", "im_cluster_rpc_duration_seconds", "im_cluster_route_total", + "im_kafka_message_persist_total", "im_kafka_consumer_lag", "im_kafka_dead_letters_total"} { + if !strings.Contains(body, "# HELP "+metric+" ") { + t.Fatalf("metric %s was not exported", metric) + } + } } func TestMonitorHTTPRecordsRequestMetrics(t *testing.T) { diff --git a/server/tinode.conf b/server/tinode.conf index daa0ed3..9ba1926 100644 --- a/server/tinode.conf +++ b/server/tinode.conf @@ -238,6 +238,11 @@ "cluster_config": { "self": "", "nodes": [], + "proxy_to_master_buffer": 4096, + "proxy_to_master_workers": 16, + "proxy_to_master_enqueue_timeout_ms": 300, + "route_retries": 3, + "route_retry_backoff_ms": 25, "failover": { "enabled": false } diff --git a/server/topic_proxy.go b/server/topic_proxy.go index 7e9e802..9ec7b43 100644 --- a/server/topic_proxy.go +++ b/server/topic_proxy.go @@ -9,12 +9,34 @@ package main import ( "errors" "net/http" + "sync/atomic" "time" "github.com/tinode/chat/server/logs" + immetrics "github.com/tinode/chat/server/metrics" "github.com/tinode/chat/server/store/types" ) +var ( + clusterProxyOverloadLogAt atomic.Int64 + clusterProxyOverloadSuppressed atomic.Int64 +) + +func logClusterProxyRouteFailure(topic, operation string, err error) { + if !errors.Is(err, errClusterLoadExceeded) && !errors.Is(err, errClusterOverloaded) { + logs.Warn.Printf("topic proxy[%s]: route %s failed - %s", topic, operation, err) + return + } + now := time.Now().Unix() + previous := clusterProxyOverloadLogAt.Load() + if previous == now || !clusterProxyOverloadLogAt.CompareAndSwap(previous, now) { + clusterProxyOverloadSuppressed.Add(1) + return + } + logs.Warn.Printf("topic proxy overload: topic=%s operation=%s suppressed=%d", topic, operation, + clusterProxyOverloadSuppressed.Swap(0)) +} + func (t *Topic) runProxy(hub *Hub) { killTimer := time.NewTimer(time.Hour) killTimer.Stop() @@ -27,8 +49,8 @@ func (t *Topic) runProxy(hub *Hub) { msg.sess.queueOut(ErrLockedReply(msg, types.TimeNow())) } else if err := globals.cluster.routeToTopicMaster(ProxyReqJoin, msg, t.name, msg.sess); err != nil { // Response (ctrl message) will be handled when it's received via the proxy channel. - logs.Warn.Printf("proxy topic[%s]: route join request from proxy to master failed - %s", t.name, err) - msg.sess.queueOut(ErrClusterUnreachableReply(msg, types.TimeNow())) + logClusterProxyRouteFailure(t.name, "join", err) + msg.sess.queueOut(clusterRouteErrorReply(msg, err)) } if msg.sess.inflightReqs != nil { msg.sess.inflightReqs.Done() @@ -51,8 +73,8 @@ func (t *Topic) runProxy(hub *Hub) { case msg := <-t.clientMsg: // Content message intended for broadcasting to recipients if err := globals.cluster.routeToTopicMaster(ProxyReqBroadcast, msg, t.name, msg.sess); err != nil { - logs.Warn.Printf("topic proxy[%s]: route broadcast request from proxy to master failed - %s", t.name, err) - msg.sess.queueOut(ErrClusterUnreachableReply(msg, types.TimeNow())) + logClusterProxyRouteFailure(t.name, "broadcast", err) + msg.sess.queueOut(clusterRouteErrorReply(msg, err)) } case msg := <-t.serverMsg: @@ -71,8 +93,8 @@ func (t *Topic) runProxy(hub *Hub) { case msg := <-t.meta: // Request to get/set topic metadata if err := globals.cluster.routeToTopicMaster(ProxyReqMeta, msg, t.name, msg.sess); err != nil { - logs.Warn.Printf("proxy topic[%s]: route meta request from proxy to master failed - %s", t.name, err) - msg.sess.queueOut(ErrClusterUnreachableReply(msg, types.TimeNow())) + logClusterProxyRouteFailure(t.name, "meta", err) + msg.sess.queueOut(clusterRouteErrorReply(msg, err)) } case upd := <-t.supd: @@ -122,6 +144,17 @@ func (t *Topic) runProxy(hub *Hub) { } } +func clusterRouteErrorReply(msg *ClientComMessage, err error) *ServerComMessage { + if errors.Is(err, errClusterLoadExceeded) || errors.Is(err, errClusterOverloaded) { + immetrics.ClusterProxyResponseTotal.WithLabelValues(clusterPodName(), "overloaded").Inc() + reply := ErrServiceUnavailableReply(msg, types.TimeNow()) + reply.Ctrl.Text = "cluster overloaded" + return reply + } + immetrics.ClusterProxyResponseTotal.WithLabelValues(clusterPodName(), "unreachable").Inc() + return ErrClusterUnreachableReply(msg, types.TimeNow()) +} + // Takes a session leave request, forwards it to the topic master and // modifies the local state accordingly. // Returns whether the operation was successful.