gRPC - Interceptor

Interceptor

gRPC 어플리케이션을 구성할 때, 서버에서나 클라이언트에서나 remote 함수들을 실행하기에 앞서 실행해야하는 여러가지 로직들이 존재할 것이다. gRPC에서는 interceptor라 불리는 로깅이나 인증, 측정등을 위해 RPC 실행전에 선점할 수 있는 기능이 있다.

 

모든 언어에서 지원하지는 않으니 주의할 것. 예시로는 Go, Java가 있으니 둘은 당연히 지원된다.

 

gRPC는 2가지 타입의 interceptor로 분화된다. 이는 RPC 타입과 같다. unary interceptor와 stream interceptor. 

server-side와 client-side로 나뉘어서 알아보자

 

Server-side interceptor 

interceptor는 하나 혹은 2개 이상으로 구성될 수 있고 예제는 다음과 같다.

내가 만들 interceptor는 서비스로 들어오는 요청들에 대한 로깅이다.

 

package chat

import (
	"context"
	
	"google.golang.org/grpc"
	"log"
)

func LogUnaryServerInterceptor(ctx context.Context, req interface{},
	info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
    // unary 에 대한 interceptor
	log.Println(info.FullMethod)
	m, err := handler(ctx, req)
	if err != nil {
		log.Printf("error %s occurred!", err)
	}
	return m, err
}


type wrappedStream struct {
	grpc.ServerStream
}


func (w *wrappedStream) RecvMsg(m interface{}) error {
	log.Printf("Recv", m)
	return w.ServerStream.RecvMsg(m)
}


func (w *wrappedStream) SendMsg(m interface{}) error {
	log.Printf("Send" ,m)
	return w.ServerStream.SendMsg(m)
}


func newWrappedStream(s grpc.ServerStream) grpc.ServerStream {
	return &wrappedStream{s}
}


func LogServerStreamInterceptor(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
	// stream에 대한 interceptor로 recv, send 모두 interceptor를 걸 수 있다.
    log.Println(info.FullMethod)
	err := handler(srv, newWrappedStream(ss))
	if err != nil {
		log.Printf("error %s occurred!", err)
	}
	return err
}

 

func main() {
	lis, err := net.Listen("tcp", ":4317")
	if err != nil {
		log.Fatalf("failed to listen: %v", err)
	}
	s := grpc.NewServer(grpc.UnaryInterceptor(chat.LogUnaryServerInterceptor),
						grpc.StreamInterceptor(chat.LogServerStreamInterceptor))
    // 이와 같이 여러개의 interceptor를 등록할 수 있다.
	pb.RegisterChatTaskServer(s, chat.InitChatServer())
	if err := s.Serve(lis); err != nil {
		log.Fatalf("failed to serve: %v", err)
	}
}

 

 

Client-side interceptor 

클라이언트 interceptor도 서버와 동일하다

 

func LogClientStreamInterceptor( ctx context.Context, desc *grpc.StreamDesc,
	cc *grpc.ClientConn, method string,
	streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
	log.Println(method)
	s, err := streamer(ctx, desc, cc, method, opts...)
	if err != nil {
		return nil, err
	}
	return newWrappedStream(s), nil
}


type wrappedStream struct {
	grpc.ClientStream
}

func (w *wrappedStream) RecvMsg(m interface{}) error {
	log.Printf("Receive a message (Type: %T)",
		m)
	return w.ClientStream.RecvMsg(m)
}

func (w *wrappedStream) SendMsg(m interface{}) error {
	log.Printf("Send a message (Type: %T)",
		m)
	return w.ClientStream.SendMsg(m)
}

func newWrappedStream(s grpc.ClientStream) grpc.ClientStream {
	return &wrappedStream{s}
}


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

	err := invoker(ctx, method, req, reply, cc, opts...)

	log.Println(reply)

	return err
}

 

 

func main() {
	conn, err := grpc.Dial(address, grpc.WithInsecure(), grpc.WithBlock(),
		grpc.WithUnaryInterceptor(LogUnaryClientInterceptor),
		grpc.WithStreamInterceptor(LogClientStreamInterceptor))
	if err != nil {
		log.Fatalf("did not connect: %v", err)
	}
	defer conn.Close()
	c := pb.NewChatTaskClient(conn)
	client := Client{ChatTaskClient: c}
	client.session()
}