如何为Go编写的长连接gRPC服务按流统计带宽并导出至Prometheus

我之前在做类似的gRPC长流服务监控时也遇到过一模一样的困扰，标准拦截器拿不到序列化后消息大小、官方中间件又没覆盖带宽指标，这里给你几个可行的实操方案：

核心思路：自定义流包装器拦截消息收发 #

gRPC的ServerStream本质是通过SendMsg和RecvMsg方法处理消息的，我们可以通过包装这个流对象，在消息实际发送/接收前后计算序列化后的字节大小，再把数据上报到Prometheus。

步骤1：定义Prometheus指标 #

首先要创建对应的监控指标，推荐用直方图（Histogram）来统计消息大小的分布，方便后续计算带宽（总字节数/时间）：

import (
    "github.com/prometheus/client_golang/prometheus"
    "github.com/prometheus/client_golang/prometheus/promauto"
)

var (
    grpcStreamMsgSize = promauto.NewHistogramVec(
        prometheus.HistogramOpts{
            Name:    "grpc_stream_message_size_bytes",
            Help:    "Distribution of gRPC stream message sizes (send/recv)",
            Buckets: prometheus.ExponentialBuckets(64, 2, 10), // 覆盖64B到32KB的常见消息大小
        },
        []string{"method", "direction", "stream_id"}, // stream_id用来区分单个长流
    )
)

步骤2：实现自定义流包装器 #

我们需要包装grpc.ServerStream，重写SendMsg和RecvMsg方法，在其中插入消息大小计算逻辑：

import (
    "google.golang.org/grpc"
    "google.golang.org/protobuf/proto"
    "github.com/google/uuid"
    "log"
)

type wrappedStream struct {
    grpc.ServerStream
    method   string // 记录gRPC方法名
    streamID string // 唯一标识单个长流
}

// 拦截发送消息，计算大小并上报
func (w *wrappedStream) SendMsg(m interface{}) error {
    // 用protobuf的Size方法计算序列化后大小（如果用其他序列化方式，替换成对应逻辑）
    size, err := proto.Size(m)
    if err != nil {
        log.Printf("failed to calculate send message size: %v", err)
    } else {
        grpcStreamMsgSize.WithLabelValues(w.method, "send", w.streamID).Observe(float64(size))
    }
    // 调用原始的SendMsg方法
    return w.ServerStream.SendMsg(m)
}

// 拦截接收消息，计算大小并上报
func (w *wrappedStream) RecvMsg(m interface{}) error {
    // 先接收消息
    err := w.ServerStream.RecvMsg(m)
    if err != nil {
        return err
    }
    // 计算接收消息的大小
    size, err := proto.Size(m)
    if err != nil {
        log.Printf("failed to calculate recv message size: %v", err)
    } else {
        grpcStreamMsgSize.WithLabelValues(w.method, "recv", w.streamID).Observe(float64(size))
    }
    return nil
}

步骤3：实现流拦截器并注册到gRPC服务 #

最后把包装器整合到流拦截器中，在服务启动时注册：

func StreamMetricsInterceptor() grpc.StreamServerInterceptor {
    return func(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
        // 生成唯一的stream ID，用UUID保证唯一性
        streamID := uuid.NewString()
        // 创建包装后的流对象
        wrapped := &wrappedStream{
            ServerStream: ss,
            method:       info.FullMethod,
            streamID:     streamID,
        }
        // 调用原始的流处理逻辑
        return handler(srv, wrapped)
    }
}

// 启动gRPC服务时添加拦截器
func main() {
    server := grpc.NewServer(
        grpc.StreamInterceptor(StreamMetricsInterceptor()),
        // 其他服务配置...
    )
    // 注册你的gRPC服务...
}

额外优化建议 #

1. 简化指标基数：如果不需要区分单个流，可以去掉stream_id标签，降低Prometheus的存储压力；
2. 非Protobuf场景适配：如果你的消息用JSON或其他序列化方式，把proto.Size(m)替换成对应序列化后的字节长度计算（比如json.Marshal(m)后取len()）；
3. 带宽计算：有了消息大小的直方图，你可以在Prometheus中通过sum(rate(grpc_stream_message_size_bytes_sum[5m]))来计算5分钟内的平均带宽。

内容的提问来源于stack exchange，提问作者Ilia Choly