Creating and testing gRPC server interceptors in Ruby
Table of Contents
If your experience is rooted in HTTP-land, then you are probably familiar with the concept of server middleware. Interceptors are analogous, but in gRPC land. When first building gRPC interceptors, I had trouble figuring out how to do it in Ruby. gRPC is not that widely used in the Ruby ecosystem, and there are not many resources on how to properly do it.
Also, my experience with writing unit and integration tests for interceptors were that there is even less documentation on the topic. I was able to find almost no how-tos and tutorials, so the best bet was reading some open source code and putting pieces together from different repositories to come up with a sane approach to it.
The lack of documentation and tutorials made me document my experience and everything I have learned in this article.
Let’s see how we can build an interceptor in Ruby and how we can test it, too.
Building the logging payload #
Legit logging interceptors or middleware usually allow developers to set the format and the logger as they wish. While this is undoubtedly the best approach, we will only look into building an interceptor that logs in JSON format for our exercise here.
To get any JSON logs out, we will need to extract some information from the gRPC call during the execution of the interceptor. To avoid polluting the interceptor with code that builds the log payload, let’s first look at a class that will have that responsibility:
module Grpclog
class Payload
CODES = {
::GRPC::Core::StatusCodes::OK => 'OK',
::GRPC::Core::StatusCodes::CANCELLED => 'Canceled',
::GRPC::Core::StatusCodes::UNKNOWN => 'Unknown',
::GRPC::Core::StatusCodes::INVALID_ARGUMENT => 'InvalidArgument',
::GRPC::Core::StatusCodes::DEADLINE_EXCEEDED => 'DeadlineExceeded',
::GRPC::Core::StatusCodes::NOT_FOUND => 'NotFound',
::GRPC::Core::StatusCodes::ALREADY_EXISTS => 'AlreadyExists',
::GRPC::Core::StatusCodes::PERMISSION_DENIED => 'PermissionDenied',
::GRPC::Core::StatusCodes::RESOURCE_EXHAUSTED => 'ResourceExhausted',
::GRPC::Core::StatusCodes::FAILED_PRECONDITION => 'FailedPrecondition',
::GRPC::Core::StatusCodes::ABORTED => 'Aborted',
::GRPC::Core::StatusCodes::OUT_OF_RANGE => 'OutOfRange',
::GRPC::Core::StatusCodes::UNIMPLEMENTED => 'Unimplemented',
::GRPC::Core::StatusCodes::INTERNAL => 'Internal',
::GRPC::Core::StatusCodes::UNAVAILABLE => 'Unavailable',
::GRPC::Core::StatusCodes::DATA_LOSS => 'DataLoss',
::GRPC::Core::StatusCodes::UNAUTHENTICATED => 'Unauthenticated'
}.freeze
attr_accessor :exception
attr_reader :method, :service, :code, :start_time
def initialize(method, code, start_time)
@service = service_name(method)
@method = method_name(method)
@code = code
@start_time = start_time
end
def to_h
result = {
'grpc.service' => service,
'grpc.method' => method,
'grpc.code' => CODES.fetch(code, code.to_s),
'grpc.start_time' => start_time.utc,
'grpc.time_ms' => elapsed_milliseconds,
'pid' => Process.pid
}
result.merge!('exception' => exception) if exception
result
end
private
def elapsed_milliseconds
(Time.now - start_time) * 1000.0
end
def method_name(method)
owner = method.owner
method_name, = owner.rpc_descs.find do |k, _|
::GRPC::GenericService.underscore(k.to_s) == method.name.to_s
end
return '(unknown)' if method_name.nil?
method_name.to_s
end
def service_name(method)
method.owner.service_name
end
end
end
Payload is only a wrapper around the gRPC method object, the start time of
the request, and the gRPC response code. Additionally, it can take an
exception if our interceptor hits an error of sorts.
Lastly, in its to_h method, the Paylaod object creates a hash that we can
use to send to our logger. Using this approach, all of the complexity related
to extracting relevant data for logging responsibility of the Payload class,
while the interceptor has to only care about logging the payload.
This way, we have separation of concerns, and each entity has single responsibility.
Logging Interceptor #
Say we have an application that speaks gRPC, and we want each inbound request to emit a logline formatted as a valid JSON. You can imagine that we can send the logger’s output to STDOUT, and we can export it and index it for further exploration.
A straightforward implementation of a server interceptor:
# lib/grpclog/server_interceptor.rb
module Grpclog
class ServerInterceptor < ::GRPC::ServerInterceptor
def initialize(logger, level = :info)
@logger = logger
@level = level
super()
end
def request_response(_request: nil, call: nil, method: nil, &block)
log(method, call, &block)
end
def server_streamer(_request: nil, call: nil, method: nil, &block)
log(method, call, &block)
end
def client_streamer(call: nil, method: nil, &block)
log(method, call, &block)
end
def bidi_streamer(_request: nil, call: nil, method: nil, &block)
log(method, call, &block)
end
end
end
The interceptor will implement all four gRPC method types, where each of them
will invoke the same method: log. The log method will take the method name
and the call object as arguments, and the gRPC method will yield to the
log method.
Let’s look quickly at the log method:
def log(method, _call)
start_time = Time.now
code = ::GRPC::Core::StatusCodes::OK
yield
rescue StandardError => e
code = e.is_a?(::GRPC::BadStatus) ? e.code : ::GRPC::Core::StatusCodes::UNKNOWN
raise
ensure
payload = Grpclog::Payload.new(method, code, start_time)
if e
payload.exception = e.message
@level = :error
end
@logger.formatter = method(:formatter)
@logger.send(@level, payload.to_h)
end
While the method is a bit busy, it doesn’t do much:
- It records the
starttime of the request, so later we can see the time spent during the call, and we can log it - It
rescues any error, to find out the error type so we can log the error type - It
ensures to create aPayload, which is a PORO containing the data we want to log that packs it all in a hash, and then logging it using the interceptor’s logger
We can see here how the Payload class is incorporated – by adding the new
object, we hide all of the complexity to building the log payload, and at the
end, we invoke the to_h method.
Testing the Payload class is straightforward; therefore, we will skip its
test for brevity’s sake.
Let’s look at how we can test our server interceptor.
Testing interceptors #
To test our interceptor, we need to think about the various types of tests:
- Unit tests, where we invoke the methods in the test using doubles for dependencies, having isolation from the rest of the gRPC stack
- Integration tests, where we mount the request on a test gRPC server, and we send requests via the network
Before we begin, a fair warning: testing interceptors in Ruby is messy (as you are about to find out). If you have figured out a cleaner approach, please drop me a line and let me know!
Unit testing intercetoprs #
When looking at the implementation of the interceptors, we can notice that to do any testing, we will need to mock out all gRPC-related dependencies:
module Grpclog
class ServerInterceptor < ::GRPC::ServerInterceptor
# Snipped for brevity
def request_response(request: nil, call: nil, method: nil, &block)
log(method, call, &block)
end
# Snipped for brevity
end
end
The request, call, and method objects are dependencies that the
interceptor’s interface forces upon us during testing. Even though our log
method does not use the request object, we will have to satisfy the interface
of the interceptor.
Additionally, if we look at the log method, we will see that we do not use
the _call object, but the method extensively uses the method object and the
Payload object that we will create in it:
def log(method, _call)
# Snipped for brevity
yield
rescue StandardError => e
# Snipped for brevity
ensure
payload = Grpclog::Payload.new(method, code, start_time)
# Snipped for brevity
@logger.send(@level, payload.to_h)
end
And the Payload#method_name and Payload#service_name:
# lib/grpclog/payload.rb
def method_name(method)
owner = method.owner
method_name, = owner.rpc_descs.find do |k, _|
::GRPC::GenericService.underscore(k.to_s) == method.name.to_s
end
return '(unknown)' if method_name.nil?
method_name.to_s
end
def service_name(method)
method.owner.service_name
end
Given the extensive use of the method object, it’s clear that a simple
double won’t do the trick.
Additionally, to unit test the service in isolation without extra dependencies, we would also have to define a gRPC service stub and implement it. That would allow us to have the test self-contained without the need for additional support files.
Without kicking the can down the road, let’s first see the setup that we will need for our unit tests:
RSpec.describe Grpclog::ServerInterceptor do
let(:rpc_class) {
Class.new do
include GRPC::GenericService
self.marshal_class_method = :encode
self.unmarshal_class_method = :decode
self.service_name = 'test.Test'
rpc :Greet, Google::Protobuf::StringValue, Google::Protobuf::Empty
end
}
let(:service_class) {
Class.new(rpc_class) do
def greet(_msg, _call)
# Do nothing
end
end
}
# Snipped for brevity
end
First, we will define the gRPC generic service class and the service class that
implements the generic class. The service will have only one method – Greet
- which will take a simple string value and return an empty value (i.e.,
Protobuf defined as Google::Protobuf::Empty).
The actual implementation of the greet method will be empty – in the spec, we
will be testing the interceptor, not the real method, so we don’t care about
its implementation.
Now that we have these classes in place, let’s continue with the test setup:
RSpec.describe Grpclog::ServerInterceptor do
# Snipped for brevity
let(:method) { service_class.new.method(:greet) }
let(:request) { double }
let(:call) { double(:call, peer: "", metadata: {}) }
# Snipped for brevity
end
We have to create a few values that we will use in the specs:
- The gRPC
methodobject - The gRPC
request– a blank double - The gRPC
call– a double withpeerandmetadataattributes
We do not need to set peer and metadata to any values for our specs.
However, keep in mind that if your interceptor accesses these attributes, like
if the interceptor extracts the User-agent from the call, you must set these
attributes on the call object.
Next, let’s look at the remainder of the spec setup:
RSpec.describe Grpclog::ServerInterceptor do
# Snipped for brevity
let(:interceptor) { described_class.new(logger) }
subject(:logger) { Logger.new(STDOUT) }
# Snipped for brevity
end
Lastly, we create the interceptor object, and pass a logger to the
constructor. The subject of our specs will be the logger object, as we
will be asserting whether the interceptor invokes the correct methods on the
logger (as seen in the implementation of the interceptor).
Let’s finally look at the actual specs:
RSpec.describe Grpclog::ServerInterceptor do
# Snipped for brevity
describe '#request_response_method' do
before { allow(subject).to receive(:send) }
context 'when no exception occurs' do
before { interceptor.request_response(request: request, call: call, method: method) { } }
it { is_expected.to have_received(:send).with(:info, Hash) }
end
context 'when a known exception occurs' do
before do
expect do
interceptor.request_response(request: request, call: call, method: method) do
raise GRPC::NotFound.new
end
end.to raise_error(GRPC::NotFound)
end
it { is_expected.to have_received(:send).with(:error, Hash) }
end
context 'when an unknown exception occurs' do
before do
expect do
interceptor.request_response(request: request, call: call, method: method) { raise :unknown }
end.to raise_error(StandardError)
end
it { is_expected.to have_received(:send).with(:error, Hash) }
end
end
# Snipped for brevity
end
We are looking at three different tests here:
- Test the logging behavior when no errors occur
- Test the behavior when a known error occurs, and
- Test the behavior when an unknown error occurs
The core of the test is the expectation that the interceptor will call a method
on the logger object, which will effectively log a line (which is the purpose
of the interceptor).
Before we kick off the test, we set an RSpec
spy, which will
watch for any invocations of the send method on the logger object (i.e.,
the subject). This spy will allow us to assert later whether the code invoked
send on the logger object, and with what arguments, too.
We will invoke the request_response method on the interceptor, with its
required argument within each spec. In other words, the request_response
method is the unit under test here. Hence, for every spec, we call the method
in a before block.
In the first spec, we invoke the method with an empty body within the when no exception occurs context. The interceptor will take the empty block, evaluate
it and log the request line. We set the expectation that when the interceptor
hits no errors, the logger will receive send with the :info argument
(i.e., the log level) and a Hash, which is the payload:
# lib/grpclog/server_interceptor.rb
@logger.send(@level, payload.to_h)
In the second spec, within the when a known exception occurs context, we use
the same approach as before with one key difference: the block evaluated by the
interceptor will raise a (known) error. In such a case, we set the expectation
that the logger will receive send with the :error argument (i.e., the log
level) and the payload Hash.
Lastly, in the third spec, within the when an unknown exception occurs
context, we follow a similar approach to the second spec, but we raise an
unknown error this time. In such a case, we set the expectation that the
logger will receive send with the :error argument (i.e., the log level)
and the payload Hash – identical to the one above.
Given that our interceptor methods perform the same functionality for all
request types, the other tests are similar to those above. To prove that the
same test approach will work for the other method, here’s the bidi_streamer
spec:
RSpec.describe Grpclog::ServerInterceptor do
# Snipped for brevity
describe '#bidi_streamer' do
before { allow(subject).to receive(:send) }
context 'when no exception occurs' do
before { interceptor.bidi_streamer(requests: [request], call: call, method: method) { } }
it { is_expected.to have_received(:send).with(:info, Hash) }
end
context 'when a known exception occurs' do
before do
expect do
interceptor.bidi_streamer(requests: [request], call: call, method: method) do
raise GRPC::NotFound.new
end
end.to raise_error(GRPC::NotFound)
end
it { is_expected.to have_received(:send).with(:error, Hash) }
end
context 'when an unknown exception occurs' do
before do
expect do
interceptor.bidi_streamer(requests: [request], call: call, method: method) { raise :unknown }
end.to raise_error(StandardError)
end
it { is_expected.to have_received(:send).with(:error, Hash) }
end
end
# Snipped for brevity
end
If we run the specs, this is the output we will get:
root@e4cdc18cf452:/app# bundle exec rspec --format documentation spec/grpclog/server_interceptor_spec.rb
Grpclog::ServerInterceptor
#request_response_method
when no exception occurs
is expected to have received send(:info, Hash) 1 time
when a known exception occurs
is expected to have received send(:error, Hash) 1 time
when an unknown exception occurs
is expected to have received send(:error, Hash) 1 time
#server_streamer
when no exception occurs
is expected to have received send(:info, Hash) 1 time
when a known exception occurs
is expected to have received send(:error, Hash) 1 time
when an unknown exception occurs
is expected to have received send(:error, Hash) 1 time
#client_streamer
when no exception occurs
is expected to have received send(:info, Hash) 1 time
when a known exception occurs
is expected to have received send(:error, Hash) 1 time
when an unknown exception occurs
is expected to have received send(:error, Hash) 1 time
#bidi_streamer
when no exception occurs
is expected to have received send(:info, Hash) 1 time
when a known exception occurs
is expected to have received send(:error, Hash) 1 time
when an unknown exception occurs
is expected to have received send(:error, Hash) 1 time
Finished in 0.02193 seconds (files took 0.16171 seconds to load)
12 examples, 0 failures
One last note: while this approach works for my use case, remember that your tests can vary depending on how complicated / what kind of work your interceptors do. You might have to rely on more mocking, or you might need more (detailed) specs. Still, knowing the above setup, you are on an excellent path to start unit testing your interceptors.
Integration testing interceptors #
To begin integration testing the interceptor, we first need to add some supporting infrastructure. We first need to define a dummy gRPC service that will have an RPC method for each method type:
- Unary
- Server streaming
- Client streaming
- Bi-directional streaming
After the service is defined, we need to implement a skeleton service to return a simple response to the RPC call. Once these pieces are in place, we can mount our interceptor to the implementation and test that the server will correctly invoke it.
Service & message definitions #
First, let’s look at the Protobuf definition of the dummy service and its message:
// spec/support/greeter.proto
syntax = "proto3";
package grpclog;
service Greeter {
rpc RequestResponseMethod(Hello) returns (Hello) {}
rpc ServerStreamMethod(Hello) returns (stream Hello) {}
rpc ClientStreamMethod(stream Hello) returns (Hello) {}
rpc BidiStreamMethod(stream Hello) returns (stream Hello) {}
}
message Hello {
string name = 1;
int64 error_code = 2;
}
Each of the RPCs defined on the Greeter service will map to the respective
method type, as supported by gRPC.
Next, we can generate the server and client stubs from the Protobuf code. Let’s
define a small rake command to aid us with this, as part of the project’s
Rakefile:
# Rakefile
namespace :test do
desc 'Generate test protobuf stubs'
task :generate_proto do |_task, _args|
system 'bundle exec grpc_tools_ruby_protoc --ruby_out=. --grpc_out=. spec/support/greeter/greeter.proto'
end
end
Now that we have the rake test:generate_proto task, we can use it to generate
the files (output shortened for brevity):
root@d20bd4fd8613:/app# rake test:generate_proto
root@d20bd4fd8613:/app# ls -la spec/support/greeter/
-rw-r--r-- 1 root root 353 Nov 1 10:50 greeter.proto
-rw-r--r-- 1 root root 482 Nov 2 20:54 greeter_pb.rb
-rw-r--r-- 1 root root 739 Nov 2 20:54 greeter_services_pb.rb
To be able to mount our server interceptor and test it, we will need to
implement an actual server, using the newly created Greeter service. The
server will have two parts: the service itself, with all its methods, and a
controller piece which will be the actual gRPC over HTTP/2 server.
Service implementation #
First, let’s put together the service methods:
# spec/support/greeter/server.rb
module Grpclog
module Greeter
class Server < Grpclog::Greeter::Service
def request_response_method(msg, _call)
raise_exception(msg)
Grpclog::Hello.new(name: "Hello #{msg.name}")
end
def server_stream_method(msg, _call)
raise_exception(msg)
[Grpclog::Hello.new(name: 'Hello!')]
end
def client_stream_method(call)
call.each_remote_read do |msg|
raise_exception(msg)
end
Grpclog::Hello.new(name: 'Hello!')
end
def bidi_stream_method(call, _view)
call.each do |msg|
raise_exception(msg)
end
[Grpclog::Hello.new(name: 'Hello!')]
end
def raise_exception(msg)
code = msg.error_code
raise ::GRPC::BadStatus.new_status_exception(code, 'test exception') if code > ::GRPC::Core::StatusCodes::OK
raise code.to_s if code < ::GRPC::Core::StatusCodes::OK
end
end
end
end
The Server implements the four methods, as defined in the Profobuf
definition. We keep the implementations lightweight, where each RPC method will
consume the request/stream and return a blank response. We leave the server as
a “shell” because we will use it to mount our interceptor for testing and only
that.
The raise_exception method is there to help us with raising an exception
during testing. During testing, it allows us to send an error_code in the
request and make the Server raise the exception. While we wouldn’t implement
such a method in a real server, this is a neat trick for testing purposes.
Now, to boot this server during testing, we will need a gRPC server implementation.
gRPC Server Implementation #
Let’s introduce a Controller class, which will do a few things:
- Initialize a new gRPC server object, with a host and a port
- Set up the gRPC server: mount the service and add the interceptors
- Provide an interface to start the server by kicking off a server thread
- Provide an interface to stop the server by terminating the server thread
Let’s quickly see the Controller:
# spec/support/greeter/controller.rb
# rubocop:disable Style/ClassVars
module Grpclog
module Greeter
class Controller
@@port = 0
class << self
def next_port
@@port += 1
end
end
attr_reader :host
def initialize(interceptors: [], service: Server)
@host = "0.0.0.0:8008#{self.class.next_port}"
@server = GRPC::RpcServer.new(
poll_period: 1,
pool_keep_alive: 1,
interceptors: interceptors
)
@server.add_http2_port(host, :this_port_is_insecure)
@server.handle(service)
end
def start
@server_thread = Thread.new { @server.run_till_terminated }
end
def stop
@server.stop
@server_thread.join
@server_thread.terminate
end
end
end
end
# rubocop:enable all
We first establish a GRPC::RpcServer during the initialization phase, the
actual gRPC server implementation, as provided by the grpc gem. We then set
up the gRPC server by attaching a host:port that it will listen to, and we
specify that the HTTP/2 connection will be insecure by using the
:this_port_is_insecure option.
When setting the port, we use a small trick. Namely, we have a next_port
class method that keeps track of the number of servers we run concurrently.
Then, every time we launch a new Controller, we increment the port number by
one. This trick allows us to spin up multiple instances of the Controller
without them trying to reuse the same port, giving us the ability to run our
tests in parallel.
(The trick is blatantly borrowed (or, rather, stolen) from the
ruby-grpc-opentracing
gem.)
In the start method, we create a new Thread and run the server. In the
stop method, we shut down the GRPC::RpcServer instance, and we also
terminate the server Thread that we start in the start method.
Next, we need the test set up to plug in our new interceptor and test it.
Integration test setup #
We begin the test setup similarly to the unit test one, but with some crucial differences:
RSpec.describe Grpclog::ServerInterceptor do
let(:log) { StringIO.new }
let(:logger) { Logger.new(log) }
let(:interceptor) { described_class.new(logger) }
let(:channel) {
GRPC::Core::Channel.new(@server.host, nil, :this_channel_is_insecure)
}
let(:client) do
Grpclog::Greeter::Stub.new(
@server.host,
:this_channel_is_insecure,
channel_override: channel,
interceptors: []
)
end
subject(:message) { JSON.parse(log.string) }
# Snipped for brevity
end
Setting up the interceptor object by passing in a logger that takes a log
which is a StringIO, is very similar to what we saw before. Next, we create a
gRPC channel that connects to a gRPC server on a specified host and port.
We use the new channel to establish a client object, which uses the
generated Grpclog::Greeter::Stub class that we saw before. We will use the
client in our specs to send actual gRPC requests to the gRPC server we
implemented.
Lastly, the subject is the parsed JSON from the log buffer. We will use it
to set expectations that the interceptor logged our messages correctly.
Let’s continue with our spec:
RSpec.describe Grpclog::ServerInterceptor do
# Snipped for brevity
before do
@server = Grpclog::Greeter::Controller.new(interceptors: [interceptor])
@server.start
end
after do
@server&.stop
end
# Snipped for brevity
end
Here, we set the before and after blocks. In the before block, we
initialize the new gRPC server by using the Grpclog::Greeter::Controller
class. To it, we pass the interceptor object that we initialized before.
In the after block, we make sure to shut down the gRPC server if booted. We
want to perform this cleanup, so we do not leave gRPC server threads running
after our specs finish.
Let’s look at a spec next:
RSpec.describe Grpclog::ServerInterceptor do
# Snipped for brevity
subject(:logger) { Logger.new(STDOUT) }
describe '#server_streamer_method' do
context 'when no exceptions are raised' do
before do
enumerator = client.server_streamer_method(Grpclog::Hello.new)
enumerator.each {} # Consume the stream
end
it { is_expected.to include('grpc.code' => 'OK') }
end
context 'when an exception is raised' do
before do
expect do
enumerator = client.server_streamer_method(
Grpclog::Hello.new(error_code: ::GRPC::Core::StatusCodes::UNKNOWN)
)
enumerator.each {} # Consume the stream
end.to raise_error(::GRPC::Unknown)
end
it { is_expected.to include('grpc.code' => 'Unknown') }
end
end
# Snipped for brevity
end
We are looking at two different specs here: one where no exception is raised
and one that raises a GRPC::Unknown.
In the first spec, we use the service client to invoke the
server_streamer_method, passing in the argument the body of the request.
Since its a streaming RPC, the returned value of the client call is an
enumerator that allows us to consume the stream.
We consume the stream returned by the server, just by looping through the
enumerator without acting on the response data itself. Next, we simply assert
on the logged line expecting the logged grpc.code field to be OK.
On the next spec we use the trick allowing us to return an error from the
server. As part of the request body, we send the error_code set as
GRPC::Core::StatusCodes::UNKNOWN, which will make the server return the same
error. The rest of the test setup is like the one before it: we consume the
stream returned by the server, just by looping through the enumerator without
acting on the response data itself.
When it comes to the expectation itself, instead of asserting that the
grpc.code be OK, we assert that the code is the error we passed as the
error_code in the request - Unknown.
If we run the specs, we will see that they pass successfully:
root@e4cdc18cf452:/app# bundle exec rspec spec/grpclog/integration/server_interceptor_spec.rb
Grpclog::ServerInterceptor
#server_streamer_method
when no exceptions are raised
is expected to include {"grpc.code" => "OK"}
when an exception is raised
is expected to include {"grpc.code" => "Unknown"}
Finished in 0.06014 seconds (files took 0.19803 seconds to load)
2 examples, 0 failures
Evidently, the two types of specs are similar - this is mainly due to the nature of the interceptor under test. On the other hand, the setup for the two types of tests is different, with the integration specs being more involved and requiring additional support files.
Still, the overhead can be worth it (depending on your case) as mounting the interceptor on an actual gRPC server gives us the confidence that our server interceptor will work correctly when mounted on a gRPC server in the wild.
If you would like to inspect the code for this article, head over to its repository. You can check out the implementation, and inspect the full list of specs, including the project setup.