4. gRPC 流量录制：拦截器、Metadata 与动态解析

gRPC 流量录制是回放工程中最具挑战性的环节之一。与 HTTP 1.x 的请求-响应模型不同，gRPC 基于 HTTP/2 帧、Protobuf 二进制序列化，并支持 Unary（一元）和 Stream（流式）两种通信模式。本章将深入讲解如何通过 gRPC 拦截器实现无侵入录制，涵盖 Metadata 与 Message 的捕获、Protobuf 动态解析以及流式数据的分片存储策略。

4.1 录制架构总览

录制层位于 gRPC 服务端/客户端拦截器链中，通过拦截 Invoke（Unary）和 RecvMsg/SendMsg（Stream）方法，将请求与响应数据异步写入存储后端。整体架构如下图所示：

4.2 Unary 拦截器录制实现

Unary 模式是最简单的 gRPC 调用形式：客户端发送一个请求，服务端返回一个响应。拦截器通过包装 UnaryServerInterceptor 或 UnaryClientInterceptor 实现录制。

4.2.1 服务端 Unary 拦截器

func UnaryRecordingInterceptor(recorder *Recorder) grpc.UnaryServerInterceptor {
    return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo,
        handler grpc.UnaryHandler) (interface{}, error) {

        // 1. 提取 Metadata
        md, _ := metadata.FromIncomingContext(ctx)
        traceID := extractTraceID(md)

        // 2. 序列化请求消息
        reqBytes, _ := proto.Marshal(req.(proto.Message))

        // 3. 调用实际 handler
        resp, err := handler(ctx, req)

        // 4. 序列化响应消息
        var respBytes []byte
        if err == nil {
            respBytes, _ = proto.Marshal(resp.(proto.Message))
        }

        // 5. 异步录制
        recorder.RecordAsync(&Record{
            TraceID:    traceID,
            Method:     info.FullMethod,
            ReqBytes:   reqBytes,
            RespBytes:  respBytes,
            Err:        err,
            Timestamp:  time.Now(),
        })

        return resp, err
    }
}

4.2.2 Metadata 录制内容

Metadata 中通常包含认证令牌、请求 ID、客户端版本等关键上下文。录制时应提取以下字段：

4.3 Stream 拦截器录制实现

流式 gRPC 包含三种模式：服务端流、客户端流、双向流。录制难点在于消息是持续到达的，需要处理分片和顺序。

4.3.1 流式拦截器核心逻辑

type recordingServerStream struct {
    grpc.ServerStream
    recorder *Recorder
    method   string
    traceID  string
    msgCount int64
}

func (s *recordingServerStream) RecvMsg(m interface{}) error {
    err := s.ServerStream.RecvMsg(m)
    if err == nil {
        s.msgCount++
        bytes, _ := proto.Marshal(m.(proto.Message))
        s.recorder.RecordStreamChunk(&StreamChunk{
            TraceID:   s.traceID,
            Method:    s.method,
            Direction: "recv",
            Seq:       s.msgCount,
            Data:      bytes,
            Timestamp: time.Now(),
        })
    }
    return err
}

func (s *recordingServerStream) SendMsg(m interface{}) error {
    bytes, _ := proto.Marshal(m.(proto.Message))
    s.msgCount++
    s.recorder.RecordStreamChunk(&StreamChunk{
        TraceID:   s.traceID,
        Method:    s.method,
        Direction: "send",
        Seq:       s.msgCount,
        Data:      bytes,
        Timestamp: time.Now(),
    })
    return s.ServerStream.SendMsg(m)
}

4.3.2 流式数据的分片存储

流式会话可能持续数分钟甚至数小时，产生数千条消息。不能将所有消息写入单个文件，必须按以下策略分片：

• 时间分片：每 60 秒或每 100 条消息生成一个新分片文件。
• 大小分片：单个分片文件超过 64MB 时自动滚动。
• 会话分片：每个 gRPC Stream 会话使用独立目录，目录名包含 traceID 和 method。
• 索引文件：记录分片元数据（起始 Seq、时间范围、文件路径），便于回放时按序加载。

4.4 Protobuf 动态解析

录制时无法预知所有消息类型，必须支持运行时动态解析 Protobuf 消息。核心思路是利用 protoreflect 或 dynamicpb 包，根据 .proto 文件描述符反序列化任意消息。

4.4.1 动态解析流程

1. 加载描述符：从 protoset 文件或 gRPC 反射服务获取 FileDescriptorSet。
2. 查找消息类型：根据 FullMethod 映射到输入/输出消息的 MessageDescriptor。
3. 动态创建消息：使用 dynamicpb.NewMessage(desc) 创建空消息实例。
4. 反序列化：调用 proto.Unmarshal(bytes, msg) 填充字段。
5. 转 JSON：使用 protojson.Marshal(msg) 输出可读格式，便于存储和回放。

import (
    "google.golang.org/protobuf/encoding/protojson"
    "google.golang.org/protobuf/reflect/protodesc"
    "google.golang.org/protobuf/types/dynamicpb"
    "google.golang.org/protobuf/types/descriptorpb"
)

func DynamicUnmarshal(fdSet *descriptorpb.FileDescriptorSet,
    msgName string, data []byte) ([]byte, error) {

    // 1. 构建文件描述符
    files, err := protodesc.NewFiles(fdSet)
    if err != nil {
        return nil, err
    }

    // 2. 查找消息描述符
    msgDesc, err := files.FindDescriptorByName(protoreflect.FullName(msgName))
    if err != nil {
        return nil, err
    }

    // 3. 动态创建消息
    msg := dynamicpb.NewMessage(msgDesc.(protoreflect.MessageDescriptor))

    // 4. 反序列化
    if err := proto.Unmarshal(data, msg); err != nil {
        return nil, err
    }

    // 5. 转 JSON
    return protojson.Marshal(msg)
}

4.5 录制数据格式与存储

每条录制记录应包含完整的上下文信息，以便回放时精确还原。推荐使用以下结构：

存储后端可选择本地文件系统（开发环境）或对象存储（生产环境）。文件命名规范：{traceID}_{method}_{timestamp}_{seq}.rec。

4.6 避坑指南

• 大消息 OOM：单个 gRPC 消息可能超过 4MB，录制时需设置缓冲区上限，超过阈值直接截断或跳过。
• 流式顺序错乱：异步录制可能导致分片写入顺序与消息实际顺序不一致。必须依赖 Seq 字段排序，而非文件修改时间。
• Metadata 大小写：gRPC Metadata 的 key 不区分大小写，但录制时建议统一转为小写存储，避免回放时匹配失败。
• 动态解析版本兼容：生产环境 proto 文件会迭代，录制时需记录 proto_version 字段，回放时使用对应版本的描述符。
• 拦截器执行顺序：如果同时存在认证、限流等多个拦截器，录制拦截器应放在最外层（最先执行），确保能捕获到所有请求。