• 即时通讯
• 后端

本文围绕 WuKongIM 仓库中的《Transport RPC Metrics Implementation Plan》（docs/superpowers/plans/2026-04-22-transport-rpc-metrics-implementation.md）展开，系统讲解如何在pkg/transport链路中埋入 dial、enqueue、RPC client 三类观察事件，并通过pkg/metrics与internal/app/observability.go落地为 6 组低基数（low-cardinality）Prometheus 指标。读者读完可以掌握指标命名、标签集合、结果分类（result bucket）、服务名翻译（service ID → 稳定名称）以及对应的 TDD 验证命令，并了解当前仓库中这些设计的实际落地位置。

一、设计目标：暴露节点拥塞与 RPC 失败，而不是淹没在基数里

该实现计划的目标非常明确：为集群节点直连 RPC 流量增加低基数（low-cardinality）的传输与 RPC 指标，覆盖以下四类信号：

• 目标节点拥塞（target-node congestion）：由 enqueue 队列饱和、stopped 状态体现；
• 拨号失败（dial failures）：由 dial 事件的dial_error结果体现；
• RPC 延迟（RPC latency）：由 RPC client 端到端耗时直方图体现；
• 超时/错误分类（timeout/error classification）：由 RPC client 结果的 8 类 bucket 体现。

关键词是"低基数"。集群规模的节点数是有限的、服务集合是有限的、结果分类是枚举的，因此最终指标序列数可控，可以直接用于告警与容量判断，而不会像按 channel、按用户打标签那样把指标基数打爆。

在仓库中，这一设计归属于 pkg/metrics/transport.go 中TransportMetrics的职责范围，注册于 pkg/metrics/registry.go 的New()（每个指标都附带node_id、node_name常量标签，来自注册表构造时的prometheus.Labels）。

二、总体架构：transport 只发信号，app 层负责翻译

计划给出的架构约束是基础设施级（infrastructure-level）解耦：

• pkg/transport层只负责发射target / service / kind / result / duration这些原始信号；
• internal/app/observability.go负责把 transport 观察事件映射为指标调用，并把服务 ID 翻译成稳定名称；
• 已有的 bytes / pool 指标保持原样不动。

// pkg/transport/internal/core/types.go 中定义的核心观察接口与事件载体 type Observer interface { ObserveTransport(event Event) } type Event struct { Name string // 事件名，如 "client_rpc"、"write_batch"、"pending_rpc" NodeID NodeID // 目标/对端节点 ServiceID uint16 // 关联服务 ID ServiceAlias string // 操作者可见的服务名（注册服务时设置） Kind FrameKind // 线框类型 Result string // 结果分类 Duration time.Duration // ... }

实际运行时，transport 的事件并不直接同步打到指标上，而是先进入有界、非阻塞的ObserverDrain（pkg/transport/internal/core/observer.go）：

• 默认有界队列容量defaultObserverQueueSize = 1024；
• 状态类事件（pending_rpc、peer_pool、scheduler_queue、service_queue等）按sourceID + priority + serviceID + nodeID去重合并（coalesce），保证队列饱和时仍保留最新绝对值；
• 计数类事件被聚合（aggregate）后按 10ms 周期批量冲刷；
• 队列打满时丢弃非终止事件，并累计到wukongim_transport_observer_dropped_total计数器（ObserveDropped），让观测丢失本身可观测。

这套"隔离热路径、合并状态、聚合计数"的设计，是该指标方案能在高频发送路径上存活的关键前提。

三、六组指标族与标签设计

计划（Task 1）要求新增的指标族及精确标签集如下，这些在 pkg/metrics/transport.go 中均已注册：

每个指标还会叠加注册表级的node_id、node_name常量标签（ConstLabels），用于区分是哪个本地节点发出的观测。源码中的 Help 文本与注册代码位于 pkg/metrics/transport.go：

rpcClientDuration: prometheus.NewHistogramVec(prometheus.HistogramOpts{ Name: "wukongim_transport_rpc_client_duration_seconds", Help: "End-to-end transport client latency in seconds; transport drains sample one in 32 calls.", ConstLabels: labels, Buckets: gatewayFrameDurationBuckets, }, []string{"target_node", "service"}),

几点值得注意的实现细节：

• 延迟采样而非全量观测：DurationSamples只保留 1/32 的延迟样本（见 pkg/transport/internal/core/types.go 的注释 "one-in-32 latency samples"），而计数保持精确（counter exact、sample reservoir bounded）。这在高频 RPC 路径上是延迟直方图不引入高开销的前提。
• 直方图桶复用：RPC client duration 与 dial duration 共用gatewayFrameDurationBuckets，定义在 pkg/metrics/gateway.go：{0.0005, 0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.15, 0.2, 0.25, 0.5, 1, 2.5, 5, 10, 30}（秒），最上界 30 秒覆盖严重 stall 场景（pkg/metrics/registry_test.go 中的TestLatencyHistogramBucketsCoverSevereStalls专门断言了这一点）。
• 句柄缓存：rpcClientDurationHandles、rpcClientTotalHandles、rpcInflightHandles、enqueueTotalHandles、dialTotalHandles、dialDurationHandles等sync.Map缓存稳定标签组合的 Vec 句柄，避免稳定热路径上反复做标签查找（TestTransportMetricsCachesStableLabelHandles验证了这些缓存条目，pkg/metrics/registry_test.go）。

四、拨号与入队事件的埋点规则

计划（Task 2）要求覆盖以下结果分类：

• dial success →result=ok
• dial failure →result=dial_error
• enqueue success →result=ok
• queue saturation →result=queue_full
• stopped pool/connection →result=stopped

实现规则强调"只观察、不改行为"：在acquire()中紧贴拨号调用前后发射 dial 事件；在Send()/RPC()中按writer.enqueue(...)的结果分类发射 enqueue 事件，保持既有运行时行为完全一致。

仓库中对应的拨号底层位于 pkg/transport/internal/peer/manager.go：

func (m *Manager) dialConn(ctx context.Context, nodeID core.NodeID) (*conn.Conn, error) { addr, err := m.cfg.Discovery.Resolve(nodeID) if err != nil { return nil, err } raw, err := m.cfg.Dial(ctx, "tcp", addr) if err != nil { if isContextError(err) { return nil, err } return nil, fmt.Errorf("%w: %v", core.ErrDialFailed, err) } c := conn.New(raw, conn.Config{Limits: m.cfg.Limits, Observer: m.cfg.Observer, NodeID: nodeID}, nil) c.Start() return c, nil }

其中core.ErrDialFailed = errors.New("transport: dial failed")定义于 pkg/transport/internal/core/types.go，是 dial 结果分类为dial_error的依据；同时该文件还定义了ErrTimeout、ErrCanceled、ErrQueueFull、ErrStopped等类型化错误，供 enqueue/RPC 结果分类使用。Manager.observe()统一走m.cfg.Observer.ObserveTransport(event)（pkg/transport/internal/peer/manager.go），观测事件最终进入ObserverDrain。

五、RPC 客户端事件与 inflight 生命周期

计划（Task 3）要求在Client.RPCService()层而非更底层埋点，因为只有这一层拿得到serviceID；每次调用发射一次 begin/end 生命周期，结果必须精确归入 8 类：

ok | timeout | canceled | remote_error | dial_error | queue_full | stopped | other

同时断言：调用期间 inflight 递增、结束后归零。

当前仓库的CallOwned（pkg/transport/client.go）已经实现了这一埋点：

func (c *Client) CallOwned(ctx context.Context, nodeID NodeID, shardKey uint64, pri Priority, serviceID uint16, payload OwnedBuffer) (_ []byte, err error) { if c.cfg.Observer != nil { started := time.Now() defer func() { result := "ok" if err != nil { result = "err" } if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, ErrTimeout) { result = "timeout" } if errors.Is(err, context.Canceled) || errors.Is(err, ErrCanceled) { result = "canceled" } c.cfg.Observer.ObserveTransport(Event{Name: "client_rpc", NodeID: nodeID, ServiceID: serviceID, Result: result, Duration: time.Since(started)}) }() } peerConn, err := c.peers.Acquire(ctx, nodeID, shardKey) // ... }

分类逻辑要点：

• 用errors.Is而不是错误文本匹配，保证context deadline exceeded→timeout、context canceled→canceled的判定是类型级而非字符串级；
• RemoteError的Unwrap()（pkg/transport/internal/core/types.go）把远端错误码timeout、canceled、busy、stopped解包回本地类型化错误，从而让远程错误也能进入同一套结果分类；
• RemoteErrorCodeGeneric = "remote_error"、RemoteErrorCodeServiceNotFound = "service_not_found"等常量则对应remote_errorbucket 的细分语义；
• inflight 归零由wukongim_transport_rpc_inflightGauge 的SetRPCInflight（pkg/metrics/transport.go）配合ObserveRPCClientBatch在应用层完成——计数与采样分离（ObserveRPCClientBatch(node, service, result, count, samples)，pkg/metrics/transport.go），先加精确计数、再消费采样延迟。

六、应用层映射：service ID 翻译为稳定名称

计划（Task 4）要求覆盖三类服务名映射：

• cluster services：forward、controller、managed_slot
• node services：presence、delivery_submit、delivery_push、delivery_ack、delivery_offline、conversation_facts、channel_append
• fallback：未知服务 →service_<id>

并要求：未知 ID 静默映射为service_<id>，不打印日志、不 panic；OnSend/OnReceive字节指标保持原样；目标节点只在 observability/metrics 层转换一次字符串。

应用层实现位于 internal/app/observability.go 的transportMetricsObserver.ObserveTransport，其中client_rpc分支：

case "client_rpc": o.metrics.Transport.ObserveRPCClientBatch( strconv.FormatUint(uint64(event.NodeID), 10), transportServiceEventLabel(event), transportRPCResultLabel(event.Result), transportEventCount(event), transportEventDurations(event), )

配套的翻译函数：

• transportServiceEventLabel(event)：优先使用event.ServiceAlias，为空时回退到transportServiceQueueLabel(ServiceID)，即"service_" + strconv.Itoa(int(serviceID))（internal/app/observability.go），正是计划中 fallbackservice_<id>的实际实现；
• transportRPCResultLabel(result)：空结果归一化为ok（internal/app/observability.go）。

集群侧的稳定名称来自 pkg/cluster/net/ids.go 的transportServiceAlias(serviceID)大表，例如RPCSlotForwardPropose → "slot propose"、RPCChannelAppend → "channel append"、RPCDeliveryPush → "delivery push"、RPCPresenceAuthority → "presence authority"、RPCPresenceOwner → "presence owner action"，未知 ID 返回"unknown service"。为了让客户端与服务端看到的服务标签一致，pkg/cluster/net/transport.go 还包了一层transportClientObserver：client_rpc事件在ServiceID <= 255时回填注册的服务别名后再交给下游 observer。

七、既有 bytes / pool 指标保持不动

计划反复强调：保持现有字节与连接池指标完整。仓库中对应证据：

• sent_bytes/received_bytes事件：observeBytes（pkg/transport/internal/conn/conn.go）按kind + priority上报，应用层映射为wukongim_transport_sent_bytes_total/received_bytes_total（标签msg_type）与lane_payload_bytes_total（标签direction + priority）；
• write_batch事件：observeWriteBatch（pkg/transport/internal/conn/conn.go）驱动write_batches_total、write_frames_total、write_payload_bytes_total、write_single_frame_batches_total、write_frame_limit_batches_total五组指标；
• 连接池：SetPoolConnections（pkg/metrics/transport.go）维护wukongim_transport_connections_pool_active/pool_idle（标签peer_node），并在peer_pool事件中与RuntimePressure的 peer_pool 工作线程数联动。

pending_rpc事件则进入RuntimePressure的 "transport/rpc" 池 inflight（按 source 去重聚合，见transportPendingRPCInflight），属于另一条独立的运行时压力观测线。

八、TDD 验证路径与测试命令

该计划采用"先写失败测试 → 实现 → 重跑通过"的 TDD 节奏，五组任务对应的测试文件与断言范围：

验证命令（可在仓库根目录执行）：

go test ./pkg/metrics -run 'TestSlotAndTransportMetricsTrackProposalsLeaderChangesAndRPCs|TestTransportMetricsTrackRPCClientDialAndEnqueue' -count=1 go test ./pkg/transport -run 'TestPoolObserverTracksDialLifecycle|TestPoolObserverTracksEnqueueResults|TestClientRPCServiceTracksObserverEvents|TestClientRPCServiceTracksInflightGauge' -count=1 go test ./internal/app -run 'TestTransportMetricsObserverRecordsRPCClientMetrics|TestTransportMetricsObserverMapsServiceIDToName' -count=1 go test ./pkg/cluster -run 'TestTransportLayerWiresTransportObserver' -count=1

注意：计划中的测试名（如TestTransportMetricsTrackRPCClientDialAndEnqueue、TestSlotAndTransportMetricsTrackProposalsLeaderChangesAndRPCs）当前已在 pkg/metrics/registry_test.go 与 pkg/metrics/registry_test.go 中存在，说明该方案在仓库中已落地；配套的集成验证还有TestTransportMetricsObserverAggregatesConnectionLocalGauges、TestTransportMetricsObserverRejectsOlderAbsoluteQueueState（internal/app/observability_test.go）以及TestTransportLayerWiresTransportObserver对应的集群接线测试。

九、观测使用建议与文档对齐

最后，计划（Task 5）要求同步pkg/cluster/FLOW.md中 transport/startup 观测描述，保持架构级表述。结合上述指标族，实践中可以这样用：

1. 判断目标节点拥塞：rate(wukongim_transport_enqueue_total{result="queue_full"}[1m])持续大于 0，配合wukongim_transport_rpc_inflight的抬升，是向节点限流或扩容的信号；
2. 判断拨号质量：wukongim_transport_dial_total{result="dial_error"}与wukongim_transport_dial_duration_seconds的 P95，用于发现地址解析失败、网络分区或对端拒绝连接；
3. 判断 RPC 健康度：按wukongim_transport_rpc_client_total的result维度观察timeout、canceled、remote_error占比，wukongim_transport_rpc_client_duration_seconds用于端到端延迟告警；
4. 注意基数边界：target_node受集群节点数约束、service受注册服务数约束、result是 8 值枚举，整套指标基数恒定，可直接接入 Prometheus 告警规则而无需担心时间序列爆炸。

十、总结

WuKongIM 的传输层 RPC 指标方案遵循一条清晰的链路：transport 层在有界ObserverDrain中发射低基数事件 →pkg/metrics注册 6 组带固定标签集的指标族 →internal/app将 service ID 翻译为稳定名称并做采样/计数分离 → 用户通过 Prometheus 观察节点拥塞、拨号失败、RPC 延迟与错误分类。整套设计的核心不是"多"，而是"低基数 + 精确计数 + 采样延迟 + 类型级错误分类"，这正是它适合作为集群级告警底座的原因。

• 即时通讯
• 后端