介绍一个grpc记录和重放库 vearne/grpcreply-使用篇

版权声明 本站原创文章 由 萌叔 发表
转载请注明 萌叔 | https://vearne.cc

1. 引言

gRPC协议应用的非常广泛，针对它流量的记录和重放是一个非常常见的需求。但是到目前为止社区中都没有找到合适的库来实现这个目标。
终于它来了。传送门: vearne/grpcreplay

2.介绍

vearne/grpcreplay是一个网络监控工具，可以记录您的 grpc流量(Unary RPC)，并将其用于灰度测试、压测或者流量分析。它有2大特点，使用简单，并且可以直接解析Protobuf的请求，并以JSON形式输出。

在使用vearne/grpcreplay前，你必须要知道vearne/grpcreplay的限制条件：

• 1. 支持解析Unary RPC，不支持streaming RPC
• 2. Server端必须开启反射 参考GRPC Server Reflection Protocol

3. 编译&使用

确保server所在的主机上已经安装了libpcap

安装libpcap可以参考以下命令：

Ubuntu

apt-get install libpcap-dev

Centos

yum install libpcap-devel

Mac

brew install libpcap

编译

make build

示例

server

package main

import (
    "context"
    "encoding/json"
    "github.com/grpc-ecosystem/go-grpc-middleware"
    pb "github.com/vearne/grpcreplay/example/search_proto"
    "google.golang.org/grpc"
    "google.golang.org/grpc/codes"
    "google.golang.org/grpc/reflection"
    "google.golang.org/grpc/status"
    "log"
    "net"
    "runtime/debug"
    "time"
)

const PORT = "35001"

type SearchServer struct{}

func (s SearchServer) Search(ctx context.Context, in *pb.SearchRequest) (*pb.SearchResponse, error) {
    return &pb.SearchResponse{StaffID: 100, StaffName: in.StaffName}, nil
}

func (s SearchServer) CurrentTime(ctx context.Context, request *pb.TimeRequest) (*pb.TimeResponse, error) {
    return &pb.TimeResponse{CurrentTime: time.Now().Format(time.RFC3339)}, nil
}

func main() {
    opts := []grpc.ServerOption{
        //grpc.Creds(c),
        grpc_middleware.WithUnaryServerChain(
            RecoveryInterceptor,
            LoggingInterceptor,
        ),
        //grpc.HeaderTableSize(0),
        //grpc.WithDisableRetry(),
    }

    server := grpc.NewServer(opts...)
    pb.RegisterSearchServiceServer(server, &SearchServer{})

    // 注册反射服务(非常重要)
    // Register reflection service on gRPC server.
    reflection.Register(server)
    lis, err := net.Listen("tcp", ":"+PORT)
    if err != nil {
        log.Fatalf("net.Listen err: %v", err)
    }

    server.Serve(lis)
}

func LoggingInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
    log.Printf("gRPC method: %s, %v", info.FullMethod, req)
    resp, err := handler(ctx, req)
    bt, _ := json.Marshal(req)
    log.Println("body", string(bt))
    log.Printf("gRPC method: %s, %v", info.FullMethod, resp)
    return resp, err
}

func RecoveryInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
    defer func() {
        if e := recover(); e != nil {
            debug.PrintStack()
            err = status.Errorf(codes.Internal, "Panic err: %v", e)
        }
    }()

    return handler(ctx, req)
}

client

package main

import (
    "context"
    "encoding/json"
    "google.golang.org/grpc"
    "google.golang.org/grpc/codes"
    "google.golang.org/grpc/metadata"
    "google.golang.org/grpc/status"
    "log"
    "time"

    pb "github.com/vearne/grpcreplay/example/search_proto"
)

const PORT = "35001"

func main() {
    conn, err := grpc.Dial(":"+PORT, grpc.WithInsecure())
    if err != nil {
        log.Fatalf("grpc.Dial err: %v", err)
    }
    defer conn.Close()

    // add some headers
    md := metadata.New(map[string]string{
        "testkey1": "testvalue1",
        "testkey2": "testvalue2",
    })
    ctx := metadata.NewOutgoingContext(context.Background(), md)

    client := pb.NewSearchServiceClient(conn)
    for i := 0; i < 1000000; i++ {
        resp, err := client.Search(ctx,
            &pb.SearchRequest{
                StaffName: "zhangsan",
                Age:       uint32(i),
                Gender:    true,
            },
        )
        if err != nil {
            statusErr, ok := status.FromError(err)
            if ok {
                if statusErr.Code() == codes.DeadlineExceeded {
                    log.Fatalln("client.Search err: deadline")
                }
            }

            log.Fatalf("client.Search err: %v", err)
        }

        bt, _ := json.Marshal(resp)
        log.Println("resp:", string(bt))
        time.Sleep(10 * time.Second)
    }
}

启动grpcreplay

可能需要执行sudo命令获取执行权限

sudo -s

捕获35001端口的gRPC的服务请求，并输出到控制台

./grpcr --input-raw="0.0.0.0:35001" --output-stdout

输出

1 9ccd5320-4dcc-11ed-abab-5626e1cdcfe2 1665977671772240000
{"headers":{":authority":"localhost:35001",":method":"POST",":path":"/SearchService/Search",":scheme":"http","content-type":"application/grpc","te":"trailers","testkey1":"testvalue1","testkey2":"testvalue2","user-agent":"grpc-go/1.48.0"},"method":"/SearchService/Search","request":"{\"staffName\":\"zhangsan\",\"gender\":true,\"age\":405084}"}

1 9ccd537a-4dcc-11ed-abab-5626e1cdcfe2 1665977671772249000
{"headers":{":authority":"localhost:35001",":method":"POST",":path":"/SearchService/Search",":scheme":"http","content-type":"application/grpc","te":"trailers","testkey1":"testvalue1","testkey2":"testvalue2","user-agent":"grpc-go/1.48.0"},"method":"/SearchService/Search","request":"{\"staffName\":\"zhangsan\",\"gender\":true,\"age\":405085}"}

1 9ccd53ca-4dcc-11ed-abab-5626e1cdcfe2 1665977671772257000
{"headers":{":authority":"localhost:35001",":method":"POST",":path":"/SearchService/Search",":scheme":"http","content-type":"application/grpc","te":"trailers","testkey1":"testvalue1","testkey2":"testvalue2","user-agent":"grpc-go/1.48.0"},"method":"/SearchService/Search","request":"{\"staffName\":\"zhangsan\",\"gender\":true,\"age\":405086}"}

参考资料

1.动图图解！收到RST，就一定会断开TCP连接吗？
2.Hypertext Transfer Protocol Version 2 (HTTP/2)
3.HPACK: Header Compression for HTTP/2
4.google.golang.org/protobuf/reflect/protoreflect
4.Go Protobuf APIv2 动态反射 Protobuf 使用指南

---