1、gRPC Middleware
go-grpc-middleware项目地址：
https://github.com/grpc-ecosystem/go-grpc-middleware
gRPC自身只能设置一个拦截器，所有逻辑都写一起会比较乱。
本篇简单介绍go-grpc-middleware的使用，包括grpc_zap、grpc_auth和grpc_recovery。
1.1 go-grpc-middleware简介
go-grpc-middleware封装了认证（auth）, 日志（ logging）, 消息（message）, 验证（validation）, 重
试（retries） 和监控（retries）等拦截器。
1.2 安装
$ go get github.com/grpc-ecosystem/go-grpc-middleware
1.3 使用方式
import "github.com/grpc-ecosystem/go-grpc-middleware"
myServer := grpc.NewServer(grpc.StreamInterceptor(grpc_middleware.ChainStreamServer(grpc_ctxtags.StreamServerInterceptor(),grpc_opentracing.StreamServerInterceptor(),grpc_prometheus.StreamServerInterceptor,// grpc_zapgrpc_zap.StreamServerInterceptor(zapLogger),// grpc_authgrpc_auth.StreamServerInterceptor(myAuthFunction),// grpc_recoverygrpc_recovery.StreamServerInterceptor(),)),grpc.UnaryInterceptor(grpc_middleware.ChainUnaryServer(grpc_ctxtags.UnaryServerInterceptor(),grpc_opentracing.UnaryServerInterceptor(),grpc_prometheus.UnaryServerInterceptor,// grpc_zapgrpc_zap.UnaryServerInterceptor(zapLogger),// grpc_authgrpc_auth.UnaryServerInterceptor(myAuthFunction),// grpc_recoverygrpc_recovery.UnaryServerInterceptor(),)),
)
grpc.StreamInterceptor中添加流式RPC的拦截器。
grpc.UnaryInterceptor中添加简单RPC的拦截器。
1.4 grpc_zap日志记录
1.4.1 创建zap.Logger实例
func ZapInterceptor() *zap.Logger {logger, err := zap.NewDevelopment()if err != nil {log.Fatalf("failed to initialize zap logger: %v", err)}grpc_zap.ReplaceGrpcLogger(logger)return logger
}
1.4.2 把zap拦截器添加到服务端
grpcServer := grpc.NewServer(grpc.StreamInterceptor(grpc_middleware.ChainStreamServer(grpc_zap.StreamServerInterceptor(zap.ZapInterceptor()),)),grpc.UnaryInterceptor(grpc_middleware.ChainUnaryServer(grpc_zap.UnaryServerInterceptor(zap.ZapInterceptor()),)),)
1.4.3 日志分析
2022-10-10T12:33:35.283+0800 INFO zap/grpclogger.go:47 [core][Server #1
] Server created {"system": "grpc", "grpc_log": true}
2022/10/10 12:33:35 listen on 1234
2022-10-10T12:33:35.303+0800 INFO zap/grpclogger.go:47 [core][Server #1ListenSocket #2] ListenSocket created {"system": "grpc", "grpc_log": true}2022-10-10T12:33:49.716+0800 INFO zap/grpclogger.go:47 [core]CPU time info is unavailable on non-linux environments. {"s
ystem": "grpc", "grpc_log": true}
2022-10-10T12:33:49.717+0800 INFO zap/options.go:212 finished unary call with code OK {"grpc.start_time": "2022-1
0-10T12:33:49+08:00", "system": "grpc", "span.kind": "server", "grpc.service": "protos.PingPong", "grpc.method": "Ping", "grpc.code
": "OK", "grpc.time_ms": 0}
2022-10-10T12:33:49.719+0800 INFO zap/grpclogger.go:47 [transport]transport: http2Server.HandleStreams failed to read fram
e: read tcp [::1]:1234->[::1]:57918: wsarecv: An existing connection was forcibly closed by the remote host. {"system": "grpc",
"grpc_log": true}
2022-10-10T12:33:49.719+0800 INFO zap/grpclogger.go:47 [transport]transport: loopyWriter.run returning. connection error:
desc = "transport is closing" {"system": "grpc", "grpc_log": true}
各个字段代表的意思如下：
{"level": "info", // string zap log levels"msg": "finished unary call", // string log message"grpc.code": "OK", // string grpc status code"grpc.method": "Ping", / string method name"grpc.service": "mwitkow.testproto.TestService", // string full name of the called service"grpc.start_time": "2006-01-02T15:04:05Z07:00", // string RFC3339 representation of the start time"grpc.request.deadline": "2006-01-02T15:04:05Z07:00", // string RFC3339 deadline of the current request if supplied"grpc.request.value": "something", // string value on the request"grpc.time_ms": 1.345, // float32 run time of the call in ms"peer.address": {"IP": "127.0.0.1", // string IP address of calling party"Port": 60216, // int port call is coming in on"Zone": "" // string peer zone for caller},"span.kind": "server", // string client | server"system": "grpc", // string"custom_field": "custom_value", // string user defined field"custom_tags.int": 1337, // int user defined tag on the ctx"custom_tags.string": "something" // string user defined tag on the ctx
}
1.4.4 把日志写到文件中
上面日志是在控制台输出的，现在我们把日志写到文件中，修改ZapInterceptor方法。
import (grpc_zap "github.com/grpc-ecosystem/go-grpc-middleware/logging/zap""go.uber.org/zap""go.uber.org/zap/zapcore""gopkg.in/natefinch/lumberjack.v2"
)// ZapInterceptor 返回zap.logger实例(把日志写到文件中)
func ZapInterceptor() *zap.Logger {w := zapcore.AddSync(&lumberjack.Logger{Filename: "log/debug.log",MaxSize: 1024, //MBLocalTime: true,})config := zap.NewProductionEncoderConfig()config.EncodeTime = zapcore.ISO8601TimeEncodercore := zapcore.NewCore(zapcore.NewJSONEncoder(config),w,zap.NewAtomicLevel(),)logger := zap.New(core, zap.AddCaller(), zap.AddCallerSkip(1))grpc_zap.ReplaceGrpcLogger(logger)return logger
}
1.5 grpc_auth认证
go-grpc-middleware中的grpc_auth默认使用authorization认证方式，以authorization为头部，包括basic,
bearer形式等。下面介绍bearer token认证。bearer允许使用access key（如JSON Web Token (JWT)）进行访
问。
1.5.1 新建grpc_auth服务端拦截器
// TokenInfo 用户信息
type TokenInfo struct {ID stringRoles []string
}// AuthInterceptor 认证拦截器，对以authorization为头部，形式为bearer token的Token进行验证
func AuthInterceptor(ctx context.Context) (context.Context, error) {token, err := grpc_auth.AuthFromMD(ctx, "bearer")if err != nil {return nil, err}tokenInfo, err := parseToken(token)if err != nil {return nil, grpc.Errorf(codes.Unauthenticated, " %v", err)}//使用context.WithValue添加了值后，可以用Value(key)方法获取值newCtx := context.WithValue(ctx, tokenInfo.ID, tokenInfo)//log.Println(newCtx.Value(tokenInfo.ID))return newCtx, nil
}//解析token，并进行验证
func parseToken(token string) (TokenInfo, error) {var tokenInfo TokenInfoif token == "grpc.auth.token" {tokenInfo.ID = "1"tokenInfo.Roles = []string{"admin"}return tokenInfo, nil}return tokenInfo, errors.New("Token无效: bearer " + token)
}//从token中获取用户唯一标识
func userClaimFromToken(tokenInfo TokenInfo) string {return tokenInfo.ID
}
代码中的对token进行简单验证并返回模拟数据。
1.5.2 客户端请求添加bearer token
实现和上篇的自定义认证方法大同小异，gRPC 中默认定义了 PerRPCCredentials，是提供用于自定义认证的接
口，它的作用是将所需的安全认证信息添加到每个RPC方法的上下文中。其包含 2 个方法：
-
GetRequestMetadata：获取当前请求认证所需的元数据。 -
RequireTransportSecurity：是否需要基于 TLS 认证进行安全传输。
接下来我们实现这两个方法：
// Token token认证
type Token struct {Value string
}const headerAuthorize string = "authorization"// GetRequestMetadata 获取当前请求认证所需的元数据
func (t *Token) GetRequestMetadata(ctx context.Context, uri ...string) (map[string]string, error) {return map[string]string{headerAuthorize: t.Value}, nil
}// RequireTransportSecurity 是否需要基于 TLS 认证进行安全传输
func (t *Token) RequireTransportSecurity() bool {return true
}
注意：这里要以authorization为头部，和服务端对应。
发送请求时添加token：
//从输入的证书文件中为客户端构造TLS凭证creds, err := credentials.NewClientTLSFromFile("../tls/server.pem", "go-grpc-example")if err != nil {log.Fatalf("Failed to create TLS credentials %v", err)}//构建Tokentoken := auth.Token{Value: "bearer grpc.auth.token",}// 连接服务器conn, err := grpc.Dial(Address, grpc.WithTransportCredentials(creds), grpc.WithPerRPCCredentials(&token))
注意：Token中的Value的形式要以bearer token值形式。因为我们服务端使用了bearer token验证方式。
1.5.3 把grpc_auth拦截器添加到服务端
grpcServer := grpc.NewServer(cred.TLSInterceptor(),grpc.StreamInterceptor(grpc_middleware.ChainStreamServer(grpc_auth.StreamServerInterceptor(auth.AuthInterceptor),grpc_zap.StreamServerInterceptor(zap.ZapInterceptor()),)),grpc.UnaryInterceptor(grpc_middleware.ChainUnaryServer(grpc_auth.UnaryServerInterceptor(auth.AuthInterceptor),grpc_zap.UnaryServerInterceptor(zap.ZapInterceptor()),)),)
写到这里，服务端都会拦截请求并进行bearer token验证，使用bearer token是规范了与HTTP请求的对接，毕竟
gRPC也可以同时支持HTTP请求。
1.6 grpc_recovery恢复
把gRPC中的panic转成error，从而恢复程序。
1.6.1 直接把grpc_recovery拦截器添加到服务端
grpcServer := grpc.NewServer(cred.TLSInterceptor(),grpc.StreamInterceptor(grpc_middleware.ChainStreamServer(grpc_auth.StreamServerInterceptor(auth.AuthInterceptor),grpc_zap.StreamServerInterceptor(zap.ZapInterceptor()),grpc_recovery.StreamServerInterceptor,)),grpc.UnaryInterceptor(grpc_middleware.ChainUnaryServer(grpc_auth.UnaryServerInterceptor(auth.AuthInterceptor),grpc_zap.UnaryServerInterceptor(zap.ZapInterceptor()),grpc_recovery.UnaryServerInterceptor(),)),)
1.6.2 自定义错误返回
当panic时候，自定义错误码并返回。
// RecoveryInterceptor panic时返回Unknown错误吗
func RecoveryInterceptor() grpc_recovery.Option {return grpc_recovery.WithRecoveryHandler(func(p interface{}) (err error) {return grpc.Errorf(codes.Unknown, "panic triggered: %v", p)})
}
1.7 go-grpc-middleware综合使用案例
新建几个文件夹用来存放相关文件：
[root@zsx middleware]# mkdir -p server/middleware/auth
[root@zsx middleware]# mkdir -p server/middleware/cred
[root@zsx middleware]# mkdir -p server/middleware/recovery
[root@zsx middleware]# mkdir -p server/middleware/zap
1.7.1 编写protoc文件
// simple.proto
// 协议为proto3
syntax = "proto3";package proto;option go_package = "./proto";// 定义发送请求信息
message SimpleRequest {// 定义发送的参数，采用驼峰命名方式，小写加下划线，如：student_name// 参数类型 参数名 标识号(不可重复)string data = 1;
}// 定义响应信息
message SimpleResponse {// 定义接收的参数// 参数类型 参数名 标识号(不可重复)int32 code = 1;string value = 2;
}// 定义我们的服务（可定义多个服务,每个服务可定义多个接口）
service Simple {rpc Route (SimpleRequest) returns (SimpleResponse) {};
}
1.7.2 生成pb.go
$ protoc --go_out=plugins=grpc:. simple.proto
1.7.3 TLS+Token认证
// ./server/middleware/cred/cred.go
package credimport ("google.golang.org/grpc""google.golang.org/grpc/credentials""log"
)// TLSInterceptor TLS证书认证
func TLSInterceptor() grpc.ServerOption {// 从输入证书文件和密钥文件为服务端构造TLS凭证creds, err := credentials.NewServerTLSFromFile("./cert/server/server.pem", "./cert/server/server.key")if err != nil {log.Fatalf("Failed to generate credentials %v", err)}return grpc.Creds(creds)
}
1.7.4 grpc_zap日志记录
// ./server/middleware/zap/zap.go
package zapimport (grpc_zap "github.com/grpc-ecosystem/go-grpc-middleware/logging/zap""go.uber.org/zap""go.uber.org/zap/zapcore""gopkg.in/natefinch/lumberjack.v2"
)// ZapInterceptor返回zap.logger实例(把日志写到文件中)
func ZapInterceptor() *zap.Logger {w := zapcore.AddSync(&lumberjack.Logger{Filename: "log/debug.log",MaxSize: 1024, //MBLocalTime: true,})config := zap.NewProductionEncoderConfig()config.EncodeTime = zapcore.ISO8601TimeEncodercore := zapcore.NewCore(zapcore.NewJSONEncoder(config),w,zap.NewAtomicLevel(),)logger := zap.New(core, zap.AddCaller(), zap.AddCallerSkip(1))grpc_zap.ReplaceGrpcLogger(logger)return logger
}// ZapInterceptor 返回zap.logger实例(把日志输出到控制台)
// func ZapInterceptor() *zap.Logger {
// logger, err := zap.NewDevelopment()
// if err != nil {
// log.Fatalf("failed to initialize zap logger: %v", err)
// }
// grpc_zap.ReplaceGrpcLogger(logger)
// return logger
// }
1.7.5 grpc_auth认证
// ./server/middleware/auth/auth.go
package authimport ("context""errors"grpc_auth "github.com/grpc-ecosystem/go-grpc-middleware/auth""google.golang.org/grpc""google.golang.org/grpc/codes"
)// TokenInfo 用户信息
type TokenInfo struct {ID stringRoles []string
}// AuthInterceptor 认证拦截器，对以authorization为头部，形式为`bearer token`的Token进行验证
func AuthInterceptor(ctx context.Context) (context.Context, error) {token, err := grpc_auth.AuthFromMD(ctx, "bearer")if err != nil {return nil, err}tokenInfo, err := parseToken(token)if err != nil {return nil, grpc.Errorf(codes.Unauthenticated, " %v", err)}//使用context.WithValue添加了值后，可以用Value(key)方法获取值newCtx := context.WithValue(ctx, tokenInfo.ID, tokenInfo)//log.Println(newCtx.Value(tokenInfo.ID))return newCtx, nil
}//解析token，并进行验证
func parseToken(token string) (TokenInfo, error) {var tokenInfo TokenInfoif token == "grpc.auth.token" {tokenInfo.ID = "1"tokenInfo.Roles = []string{"admin"}return tokenInfo, nil}return tokenInfo, errors.New("Token无效: bearer " + token)
}//从token中获取用户唯一标识
func userClaimFromToken(tokenInfo TokenInfo) string {return tokenInfo.ID
}
1.7.6 grpc_recovery恢复
// ./server/middleware/recovery/recovery.go
package recoveryimport (grpc_recovery "github.com/grpc-ecosystem/go-grpc-middleware/recovery""google.golang.org/grpc""google.golang.org/grpc/codes"
)// RecoveryInterceptor panic时返回Unknown错误吗
func RecoveryInterceptor() grpc_recovery.Option {return grpc_recovery.WithRecoveryHandler(func(p interface{}) (err error) {return grpc.Errorf(codes.Unknown, "panic triggered: %v", p)})
}
1.7.7 Toekn
// ./token/token.go
package authimport ("context"
)// Token token认证
type Token struct {Value string
}const headerAuthorize string = "authorization"// GetRequestMetadata 获取当前请求认证所需的元数据
func (t *Token) GetRequestMetadata(ctx context.Context, uri ...string) (map[string]string, error) {return map[string]string{headerAuthorize: t.Value}, nil
}// RequireTransportSecurity 是否需要基于 TLS 认证进行安全传输
func (t *Token) RequireTransportSecurity() bool {return true
}
1.7.8 服务端
package mainimport ("context"grpc_middleware "github.com/grpc-ecosystem/go-grpc-middleware"grpc_auth "github.com/grpc-ecosystem/go-grpc-middleware/auth"grpc_zap "github.com/grpc-ecosystem/go-grpc-middleware/logging/zap"grpc_recovery "github.com/grpc-ecosystem/go-grpc-middleware/recovery"pb "middleware/proto""middleware/server/middleware/auth""middleware/server/middleware/cred""middleware/server/middleware/recovery""middleware/server/middleware/zap""google.golang.org/grpc""log""net"
)// SimpleService 定义我们的服务
type SimpleService struct{}// Route 实现Route方法
func (s *SimpleService) Route(ctx context.Context, req *pb.SimpleRequest) (*pb.SimpleResponse, error) {res := pb.SimpleResponse{Code: 200,Value: "hello " + req.Data,}return &res, nil
}const (// Address 监听地址Address string = ":8000"// Network 网络通信协议Network string = "tcp"
)func main() {// 监听本地端口listener, err := net.Listen(Network, Address)if err != nil {log.Fatalf("net.Listen err: %v", err)}// 新建gRPC服务器实例grpcServer := grpc.NewServer(// TLS+Token认证cred.TLSInterceptor(),grpc.StreamInterceptor(grpc_middleware.ChainStreamServer(// grpc_zap日志记录grpc_zap.StreamServerInterceptor(zap.ZapInterceptor()),// grpc_auth认证grpc_auth.StreamServerInterceptor(auth.AuthInterceptor),// grpc_recovery恢复grpc_recovery.StreamServerInterceptor(recovery.RecoveryInterceptor()),)),grpc.UnaryInterceptor(grpc_middleware.ChainUnaryServer(// grpc_zap日志记录grpc_zap.UnaryServerInterceptor(zap.ZapInterceptor()),// grpc_auth认证grpc_auth.UnaryServerInterceptor(auth.AuthInterceptor),// grpc_recovery恢复grpc_recovery.UnaryServerInterceptor(recovery.RecoveryInterceptor()),)),)// 在gRPC服务器注册我们的服务pb.RegisterSimpleServer(grpcServer, &SimpleService{})log.Println(Address + " net.Listing with TLS and token...")//用服务器 Serve() 方法以及我们的端口信息区实现阻塞等待，直到进程被杀死或者 Stop() 被调用err = grpcServer.Serve(listener)if err != nil {log.Fatalf("grpcServer.Serve err: %v", err)}
}
[root@zsx middleware]# go run server.go
2023/02/11 20:51:48 :8000 net.Listing with TLS and token...
1.7.9 客户端
package mainimport ("context"pb "middleware/proto""middleware/token""google.golang.org/grpc""google.golang.org/grpc/credentials""log"
)// Address 连接地址
const Address string = ":8000"var grpcClient pb.SimpleClientfunc main() {//从输入的证书文件中为客户端构造TLS凭证creds, err := credentials.NewClientTLSFromFile("./cert/server/server.pem", "test.example.com")if err != nil {log.Fatalf("Failed to create TLS credentials %v", err)}//构建Tokentoken := auth.Token{Value: "bearer grpc.auth.token",}// 连接服务器conn, err := grpc.Dial(Address, grpc.WithTransportCredentials(creds), grpc.WithPerRPCCredentials(&token))if err != nil {log.Fatalf("net.Connect err: %v", err)}defer conn.Close()// 建立gRPC连接grpcClient = pb.NewSimpleClient(conn)route()
}// route 调用服务端Route方法
func route() {// 创建发送结构体req := pb.SimpleRequest{Data: "grpc",}// 调用我们的服务(Route方法)// 同时传入了一个 context.Context ，在有需要时可以让我们改变RPC的行为，比如超时/取消一个正在运行的RPCres, err := grpcClient.Route(context.Background(), &req)if err != nil {log.Fatalf("Call Route err: %v", err)}// 打印返回值log.Println(res)
}
[root@zsx middleware]# go run client.go
2023/02/11 21:01:24 code:200 value:"hello grpc"
1.7.10 生成的日志
[root@zsx protoc]# cd middleware/
[root@zsx middleware]# cat ./log/debug.log
{"level":"info","ts":"2023-02-11T21:02:44.142+0800","caller":"grpclog/logger.go:53","msg":"[core][Server #1] Server created","system":"grpc","grpc_log":true}
{"level":"info","ts":"2023-02-11T21:02:44.142+0800","caller":"grpclog/logger.go:53","msg":"[core][Server #1 ListenSocket #2] ListenSocket created","system":"grpc","grpc_log":true}
{"level":"info","ts":"2023-02-11T21:02:49.814+0800","caller":"zap/server_interceptors.go:39","msg":"finished unary call with code OK","grpc.start_time":"2023-02-11T21:02:49+08:00","system":"grpc","span.kind":"server","grpc.service":"proto.Simple","grpc.method":"Route","grpc.code":"OK","grpc.time_ms":0.02}
{"level":"info","ts":"2023-02-11T21:02:49.815+0800","caller":"grpclog/logger.go:53","msg":"[transport]transport: closing: EOF","system":"grpc","grpc_log":true}
{"level":"info","ts":"2023-02-11T21:02:49.816+0800","caller":"grpclog/logger.go:53","msg":"[transport]transport: loopyWriter exited. Closing connection. Err: transport closed by client","system":"grpc","grpc_log":true}
# 项目结构
[root@zsx protoc]# tree middleware/
middleware/
├── cert
│ ├── ca.crt
│ ├── ca.csr
│ ├── ca.key
│ ├── ca.srl
│ ├── client
│ │ ├── client.csr
│ │ ├── client.key
│ │ └── client.pem
│ ├── openssl.cnf
│ └── server
│ ├── server.csr
│ ├── server.key
│ └── server.pem
├── client.go
├── go.mod
├── go.sum
├── log
│ └── debug.log
├── proto
│ └── simple.pb.go
├── server
│ └── middleware
│ ├── auth
│ │ └── auth.go
│ ├── cred
│ │ └── cred.go
│ ├── recovery
│ │ └── recovery.go
│ └── zap
│ └── zap.go
├── server.go
├── simple.proto
└── token└── token.go12 directories, 23 files
1.8 总结
本篇介绍了go-grpc-middleware中的grpc_zap、grpc_auth和grpc_recovery拦截器的使用。go-grpc-middleware
中其他拦截器可参考GitHub https://github.com/grpc-ecosystem/go-grpc-middleware学习使用。
本文链接：https://my.lmcjl.com/post/9641.html
展开阅读全文
4 评论