interceptor拦截器--gRPC的Middleware

[kevinyan815]

什么是拦截器

gRPC的拦截器（interceptor）类似各种Web框架里的请求中间件，请求中间件大家都知道是利用装饰器模式对最终处理请求的handler程序进行装饰，这样中间件就可以在处理请求前和完成处理后这两个时机上，拦截到发送给 handler 的请求以及 handler 返回给客户端的响应 。
中间件的最大的用处是可以把一些 handler 的前置和后置操作从 handler 程序中解耦出来，比如最常见的记录响应时长、记录请求和响应数据日志等操作往往是通过中间件程序实现的。
与 Web 框架的中间件同理，可以对gRPC的请求和响应进行拦截处理，而且既可以在客户端进行拦截，也可以对服务器端进行拦截。利用拦截器，可以对gRPC进行很好的扩展，把一些业务逻辑外的冗余操作从 handler 中抽离，提升项目的开发效率和扩展性。

怎么使用拦截器

gRPC的服务器和客户端都是分别可以添加一个单向调用 (Unary) 的拦截器和流式调用 (Stream) 的拦截器。

这两种调用方式的区别可以理解为HTTP和WebSocket的区别

对于客户端的单向调用的拦截，只需定义一个 UnaryClientInterceptor 方法：

type UnaryClientInterceptor func(ctx context.Context, method string, req, reply interface{}, cc *ClientConn, invoker UnaryInvoker, opts ...CallOption) error

而客户端流式调用的拦截，则需要定义一个 StreamClientInterceptor 方法：

type StreamClientInterceptor func(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, streamer Streamer, opts ...CallOption) (ClientStream, error)

同理，对于gRPC的服务端也有这两种调用的拦截器方法，分别是 UnaryServerInterceptor 和 StreamServerInterceptor：

type UnaryServerInterceptor func(ctx context.Context, req interface{}, info *UnaryServerInfo, handler UnaryHandler) (resp interface{}, err error)

type StreamServerInterceptor func(srv interface{}, ss ServerStream, info *StreamServerInfo, handler StreamHandler) error

拦截器应用

下面简单演示一下，怎么用客户端和服务端拦截器来实现gRPC客户端调用日志，和gRPC服务器访问日志的。

首先我们定义一下客户端单向调用的拦截器方法：

func UnaryClientInterceptor(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) (err error) {

	p := peer.Peer{}
	if opts == nil {
		opts = []grpc.CallOption{grpc.Peer(&p)}
	} else {
		opts = append(opts, grpc.Peer(&p))
	}

	start := time.Now()
	defer func() {
		in, _ := json.Marshal(req)
		out, _ := json.Marshal(reply)
		inStr, outStr := string(in), string(out)
		duration := int64(time.Since(start) / time.Millisecond)

		var remoteServer string
		if p.Addr != nil {
			remoteServer=p.Addr.String()
		}

		log.Println("grpc", method, "in", inStr, "out", outStr, "err", err, "duration/ms", duration, "remote_server", remoteServer)

	}()

	return invoker(ctx, method, req, reply, cc, opts...)
}

创建客户端的时候应用上这个方法：

var client routeGuideClient

func init() {
	var err error
	client.cc, err = grpc.Dial(
		"127.0.0.1:12305",
		grpc.WithInsecure(),
		grpc.WithUnaryInterceptor(UnaryClientInterceptor),
	)
	if err != nil {
		panic(err)
	}

}

routeguide 这个服务名是自己起的，其实就是拿的 grpc 官方的示例稍微改动了一下做的试验。

接下来定义一个服务器端的拦截器：

func UnaryServerInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
	remote, _ := peer.FromContext(ctx)
	remoteAddr := remote.Addr.String()

	in, _ := json.Marshal(req)
	inStr := string(in)
	log.Println("ip", remoteAddr, "access_start", info.FullMethod, "in", inStr)

	start := time.Now()
	defer func() {
		out, _ := json.Marshal(resp)
		outStr := string(out)
		duration := int64(time.Since(start) / time.Millisecond)
		if duration >= 500 {
			log.Println("ip", remoteAddr, "access_end", info.FullMethod, "in", inStr, "out", outStr, "err", err, "duration/ms", duration)
		} else {
			log.Println("ip", remoteAddr, "access_end", info.FullMethod, "in", inStr, "out", outStr, "err", err, "duration/ms", duration)
		}
	}()

	resp, err = handler(ctx, req)

	return
}

在服务器启动时应用上这个单向调用的拦截器：

var (
	port = flag.Int("p", 12305, "port")
)

type server struct {}

func (s server) Ping(ctx context.Context, request *routeguide.PingRequest) (reply *routeguide.PingReply, err error) {
	reply = &routeguide.PingReply{
		Reply:                "pong",
	}

	return
}

func main() {
	lis, err := net.Listen("tcp", fmt.Sprintf(":%d", *port))
	if err != nil {
		log.Fatalf("failed to listen: %v", err)
	}

	s := grpc.NewServer(grpc.UnaryInterceptor(UnaryServerInterceptor))
	routeguide.RegisterRouteGuideServer(s, &server{})
	s.Serve(lis)
}

启动服务器，用客户端调用后可以看到在服务器和客户端采集到的日志：

// 服务器日志
2021/04/24 17:24:44 ip 127.0.0.1:57258 access_start /routeguide.RouteGuide/Ping in {}
2021/04/24 17:24:44 ip 127.0.0.1:57258 access_end /routeguide.RouteGuide/Ping in {} out {"reply":"pong"} err <nil> duration/ms 0

// 客户端日志
2021/04/24 17:41:11 grpc /routeguide.RouteGuide/Ping in {} out {"reply":"pong"} err <nil> duration/ms 1 

与Web框架的中间件不同的是，Web框架可以给每个 handler 程序应用多个中间件，但是gRPC的客户端和服务器分别可以添加一个单向调用类型的拦截器和流式调用类型的拦截器。不过gRPC社区里Go gRPC Middleware 这个软件包提供了拦截器的interceptor链式的功能，可以将多个拦截器组合成一个拦截器链。

var client routeGuideClient

func init() {
	var err error
	client.cc, err = grpc.Dial(
		"127.0.0.1:12305",
		grpc.WithInsecure(),
		grpc.WithUnaryInterceptor(grpc_middleware.ChainUnaryClient(UnaryClientInterceptor)), // 参数里可以添加多个拦截器
	)
	if err != nil {
		panic(err)
	}

}

上面的示例程序为了便于理解做了部分删减，完整可运行的源码可以访问 GitHub 链接获得。

社区里那些实用的拦截器

利用拦截器，可以对gRPC进行扩展，利用社区的力量将gRPC发展壮大，也可以让开发者更灵活地处理gRPC流程中的业务逻辑。下面列出了利用拦截器实现的一些功能框架：

1. Go gRPC Middleware:提供了拦截器的interceptor链式的功能，可以将多个拦截器组合成一个拦截器链，当然它还提供了其它的功能，所以以gRPC中间件命名。
2. grpc-multi-interceptor: 是另一个interceptor链式功能的库，也可以将单向的或者流式的拦截器组合。
3. grpc_auth: 身份验证拦截器
4. grpc_ctxtags: 为上下文增加Tag map对象
5. grpc_zap: 支持zap日志框架
6. grpc_logrus: 支持logrus日志框架
7. grpc_prometheus: 支持 prometheus
8. otgrpc: 支持opentracing/zipkin
9. grpc_opentracing:支持opentracing/zipkin
10. grpc_retry: 为客户端增加重试的功能
11. grpc_validator: 为服务器端增加校验的功能