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

版权声明 本站原创文章 由 萌叔 发表
转载请注明 萌叔 | http://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 使用指南