guide to gRPC interceptors
A look at the intricacies of gRPC interceptors, how to get started with your first implementation and a real world example — deployments in on-premises locations.
This blog assumes a general understanding and competency when working with protocol buffers already — but if you’re new to this concept, there are several fantastic articles on the web that can serve as a preface to this one, such as the Go basics tutorial from the creators of gRPC here.
In this blog, we’re going to cover what gRPC interceptors are, alongside an open-source example that you can make your own changes to, and see how the different components come together to form more complex systems.
Interceptors are neat little components of a gRPC application that allow us to interact with a proto message or context either before — or after — it is sent or received by the client or server. For example, we may want to modify each request before it is sent, perhaps by adding some information or metadata about the environment the client is running in — and we can do that! On the server side — we can also intercept that message before the actual function call is executed — perhaps running some validation or a check before our main business logic is run.
There are limitless reasons why this is useful as you can probably already infer, but to list a few common use cases:
In order to not confuse things — there are four different kinds of interceptor — 2 unary interceptors, and 2 stream interceptors. One for the client and server!
These are executed on the client before a singular request is made to a function on the server side. This is where we might enrich a message with client side metadata, such as some information about the hardware or OS the client is running on, or potentially start our tracing flow.
These are executed server side when a singular request is received from a client. It’s at this point we may want to perform some checks on the authenticity of the request, such as authorising it, or checking that certain fields are present / validating the request.
These serve the same purpose as the unary interceptor, only for when we are streaming data from a client to a server. For example, if we were streaming a list of 100 objects to the server, such as chunks of a file or video, we could intercept before sending each chunk, and validating things like the checksum are valid, adding metadata to a frame and the likes.
These serve the same purpose as the server interceptor, only for when we are receiving streamed data from a client. For example, if we’re receiving the aforementioned chunks of a file, maybe we want to establish that nothing has been lost in transit and verify the checksum again before storing.
Here at Ori, we specialise in making the future of cloud computing, and right now — that’s edge. Lets say, hypothetically, that we are running a series of micro-services in our cloud. All of these services allow a customer, whether that be a local supermarket, warehouse, or user with some spare compute in his basement, to register with Ori using a corresponding client on their hardware. Once registered, we take some of their information and evaluate whether we can on-board our infrastructure onto their compute, and allow public end users to to run their containerised workloads on these nodes.
Users that wish to enroll, must install a client that allows a user to register their hardware. However — we also want to grab some metadata about the hardware that the layman user may not know, such as the operating system the client is running on, and the IP address the client is connecting from.
This is where our interceptors come in! In this case, we’re going to grab the operating system of the client and get the IP address on the server side.
Ready? Let’s Build!
You can download the source code for both the client and server as they exist from this repository: https://github.com/ori-edge/grpc-interceptor-demo
Open a terminal and simply run make protoc && make build to ensure that the generated files are up to date, and see the client and server binaries located in your bin directory within the repo.
Now, open up 2 terminals. One terminal is going to be used to run your server, whilst the other is going to be where you are tinkering with your client. To run the server, navigate to your terminal and cd to the repo directory. Run the server with ./bin/edge-server. You should see the following in your terminal, indicating that your server is running correctly:
2021/05/28 16:17:20 starting server...The server currently exposes two bits of functionality to an end-user:
--list flag.To register a client:
./bin/edge-client register
Should create the following output in the server:
2021/05/28 16:25:24 registering client success
To list clients that are registered with Ori’s network:
./bin/edge-client list --list=c45d87da-95c8-4fe7-8403-cb798e9d2805,dc441b38-851a-4c28-b82c-f1adfa271f10
Should create the following output in the server:
2021/05/28 16:28:10 streaming edge locations...
And return those locations to a user:
2021/05/28 16:28:10 id:"c45d87da-95c8-4fe7-8403-cb798e9d2805" updated_at:{seconds:1622215524 nanos:854851258}
2021/05/28 16:28:10 id:"dc441b38-851a-4c28-b82c-f1adfa271f10" updated_at:{seconds:1622215524 nanos:854851258}
Now that we have our local environment up and running, open up a code editor of your choosing — let’s build those interceptors!
The first thing we are going to do is to create a new package within our pkg/ directory, let’s call this directory interceptor and add an interceptor.go file here: this is where all of the business logic for our interceptors will live.
We are then going to define a function that will handle our unary client side interceptor. For that, lets create a function that has the following method signature:
func UnaryClientInterceptor() grpc.UnaryClientInterceptor {
}
As you can see, when this is called, we return the gRPC internal type of unary client interceptor. Let’s flesh this out!
Within this function we are going to add the following functionality:
return func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
// Get the operating system the client is running on
cos := runtime.GOOS // Append the OS info to the outgoing request
ctx = metadata.AppendToOutgoingContext(ctx, "client-os", cos) // Invoke the original method call
err := invoker(ctx, method, req, reply, cc, opts...) log.Printf("client interceptor hit: appending OS: '%v' to metadata", cos) return err
}
Breaking this down:
return func() part of this function adheres to the method signature of a unary client interceptorcosinvoker function calls the intended function call before we intercepted it, with the now modified contextWe have the functionality of our interceptor now, but how do we make it work? This is really simple! When we dial our gRPC server in the client:
conn, err := grpc.Dial("localhost:5565", grpc.WithInsecure())
We just need to add an additional option:
conn, err := grpc.Dial(
"localhost:5565",
grpc.WithInsecure(),
grpc.WithUnaryInterceptor(interceptor.UnaryClientInterceptor()),
)
To test that it works, run your client and server as detailed above, and see if you see the debug log of the interceptor being hit!
We are then going to define a function that will handle our unary server side interceptor. For that, lets create a function that has the following method signature within interceptor.go:
func UnaryServerInterceptor() grpc.UnaryServerInterceptor {
}
To flesh out this function call, add the following body to the above function:
return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
// Get the metadata from the incoming context
md, ok := metadata.FromIncomingContext(ctx)
if !ok {
return nil, fmt.Errorf("couldn't parse incoming context metadata")
} // Retrieve the client OS, this will be empty if it does not exist
os := md.Get("client-os")
// Get the client IP Address
ip, err := getClientIP(ctx)
if err != nil {
return nil, err
} // Populate the EdgeLocation type with the IP and OS
req.(*api.EdgeLocation).IpAddress = ip
req.(*api.EdgeLocation).OperatingSystem = os[0] h, err := handler(ctx, req)
log.Printf("server interceptor hit: hydrating type with OS: '%v' and IP: '%v'", os[0], ip) return h, err
}
Then, add an additional function to get the client IP address from the incoming context:
// GetClientIP inspects the context to retrieve the ip address of the client
func getClientIP(ctx context.Context) (string, error) {
p, ok := peer.FromContext(ctx)
if !ok {
return "", fmt.Errorf("couldn't parse client IP address")
} return p.Addr.String(), nil
}
Again, let’s break down what this is doing:
metadata.FromIncomingContext(ctx). This grants us access to the client-os key value pair we created earliergetClientIP functionhandler, passing any modified contexts or requests on to that function callIn order to register this with the server, add the following line when creating your new gRPC server instance:
s := grpc.NewServer()
Should become:
s := grpc.NewServer(grpc.UnaryInterceptor(interceptor.UnaryServerInterceptor()))
This functionality intercepts every streamed object from the client to the server — in this case — listing edge locations.
Create a new function in interceptor.go:
// StreamClientInterceptor allows us to log on each client stream opening
func StreamClientInterceptor() grpc.StreamClientInterceptor {
return func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
log.Printf("opening client streaming to the server method: %v", method) return streamer(ctx, desc, cc, method)
}
}
As you can see from previous implementations, we follow the exact same workflow. We don’t actually perform any business logic here, bar logging the gRPC method that is being called. However, this is the perfect use case for initialising tracing or the likes, and appending it to the outgoing context.
In order to register this stream client interceptor with our client, add to the existing dial options the following:
conn, err := grpc.Dial(
"localhost:5565",
grpc.WithInsecure(),
grpc.WithUnaryInterceptor(interceptor.UnaryClientInterceptor())
)
Should become:
conn, err := grpc.Dial(
"localhost:5565",
grpc.WithInsecure(),
grpc.WithUnaryInterceptor(interceptor.UnaryClientInterceptor()),
grpc.WithStreamInterceptor(interceptor.StreamClientInterceptor()),
)
The way a stream server interceptor works is a little more difficult to wrap your head around — we have to embed the grpc.ServerStream type in a struct, in order to get access to the RecvMsg function. For example, to create our interceptor as we have seen before:
// Set up a wrapper to allow us to access the RecvMsg function
func StreamServerInterceptor() grpc.StreamServerInterceptor {
return func(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
wrapper := &EdgeServerStream{
ServerStream: ss,
}
return handler(srv, wrapper)
}
}
This looks fairly familiar, though we are using a wrapper. What does that allow us to do? Let’s investigate further.
// Embedded EdgeServerStream to allow us to access the RecvMsg function on
// intercept
type EdgeServerStream struct {
grpc.ServerStream
}// RecvMsg receives messages from a stream
func (e *EdgeServerStream) RecvMsg(m interface{}) error {
// Here we can perform additional logic on the received message, such as
// validation
log.Printf("intercepted server stream message, type: %s", reflect.TypeOf(m).String())
if err := e.ServerStream.RecvMsg(m); err != nil {
return err
}
return nil
}
By wrapping the sever stream in our own type, we allow ourselves access to the RecvMsg function, which as above displays, allows us to interact with each message received at a time! Pretty useful, huh?
In order to register this with our server, change the following code to implement the stream server interceptor you just created:
s := grpc.NewServer(
grpc.UnaryInterceptor(interceptor.UnaryServerInterceptor()),
)s := grpc.NewServer(
grpc.UnaryInterceptor(interceptor.UnaryServerInterceptor()),
grpc.StreamInterceptor(interceptor.StreamServerInterceptor()),
)
The above has hopefully given you an example of each of the interceptors that we have available to us when working with gRPC applications with a real world application and use case that you can interact with. Remember that all of the code is available in our public GitHub repository here: https://github.com/ori-edge/grpc-interceptor-demo/. There is an open pull request on this repo where you can see the changes that are required to implement the above interceptors easily: https://github.com/ori-edge/grpc-interceptor-demo/pull/2.
Thanks for reading! Feel free to ping us a note in our community if you need a hand at all.
Discover why we opted for gRPC over REST for our open-source tool Multy. Our blog discusses the benefits and drawbacks of each protocol for modern...
Kubernetes usage continues to grow, despite its complexity
Gartner top IT trends for 2021 includes multi cloud