Interceptors

Never go out to meet trouble. If you just sit still, nine cases out of ten, someone will intercept it before it reaches you.

-- Calvin Coolidge

In the Kafka cycle, in the chapter about communication paradigms, we mentioned the pipes-and-filters pattern of communication. This pattern has data being processed, modified or enriched once or more times along the way until it reaches the destination.

When we consider web application architecture, we can find a similarity in the concept of middlewares.

Interceptors are similar to middlewares, but they are way more limited. Interceptors are transparent to the application logic, and so you can use them for many common tasks, which could have otherwise led to duplicate code.

Using interceptors can also help make implementations be clearer in the meaningful action that they do.

For example, imagine that you need to perform some data validation. You are receiving a message about a User, and it has a user id. User IDs have a particular format. Instead of validating that the format is correct in all of the gRPC calls' implementations, you can write an interceptor that does this for you.

This interceptor would validate that the ID is correct, and if it isn't, it will prevent the gRPC call from proceeding further.

You could take this even a step further and verify with the database that the user in fact exists.

If you are accessing the database, you might as well use it to validate authentication. Now, you no longer have to worry about it in each RPC call, as the interceptor would reject every unauthenticated call.

Incoming and outgoing

gRPC interceptors can be utilized by both parties of the communication. Client-side interceptors, that is interceptors for outgoing RPC calls, are slightly less useful in terms of doing actionable thing, as it is unlikely, that you would want to stop a call for any reason on your server.

However, they are fairly useful for the two following applications:

logging
metrics

Client-side interceptors are the ideal place for a logging harness that will log every call with its metadata, and also a great place to collect client-side metrics about RPC call usage.

Of course, these are both applications that you can do on the server-side also, here we would be talking about interceptors for incoming calls.

To reiterate, the usecases for incoming interceptors are:

logging
metrics
data validation
authentication

In production, your gRPC calls may end up being a pipeline with interceptors present both on the side of the client and the server:

an example with two client-side interceptors and one server-side interceptor

In Rust - Tonic specifics

Rust has multiple crates for dealing with gRPC. However, the one that is the most mature at the time of this writing, and which is also the one we use, tonic, has some specifics when dealing with interceptors.

Tonic is related to the tower project, which is a library of modular and reusable components for building robust clients and servers. Tower is protocol agnostic, however it is more suited to the request-response pattern than anything.

Interceptors are a bit more limited in tonic, they can essentially only do two things:

add/remove/check items in the MetadataMap of a request
cancel a request with a Status

These may be enough to implement functionality for the aforementioned use cases, but users are actually discouraged from doing logging and metrics with them and creating a middleware in tower is the recommended way.

As a matter of fact, tower-http has a built-in Trace middleware which already implements logging for gRPC, so you can either use that directly, or take inspiration when implementing your own logging.

https://docs.rs/tower-http/latest/tower_http/trace/index.html

#![allow(unused)]
fn main() {
use http::{Request, Response};
use hyper::Body;
use tower::{ServiceBuilder, ServiceExt, Service};
use tower_http::trace::TraceLayer;
use std::convert::Infallible;

async fn handle(request: Request<Body>) -> Result<Response<Body>, Infallible> {
    Ok(Response::new(Body::from("foo")))
}

// Setup tracing
tracing_subscriber::fmt::init();

let mut service = ServiceBuilder::new()
    .layer(TraceLayer::new_for_http())
    .service_fn(handle);

let request = Request::new(Body::from("foo"));

let response = service
    .ready()
    .await?
    .call(request)
    .await?;
}

*example of using the Trace tower middleware

To create an interceptor in tonic, you simply have to implement the Interceptor trait:

#![allow(unused)]
fn main() {
pub trait Interceptor {
    fn call(&mut self, request: Request<()>) -> Result<Request<()>, Status>;
}
}

The trait is pretty straight-forward, you take a call with the request in question, and if you return an Err(status), the call is interrupted and doesn't make it further down the pipeline.

A complete example with an interceptor might look something like this:

use hello_world::greeter_client::GreeterClient;
use hello_world::HelloRequest;
use tonic::{
    codegen::InterceptedService,
    service::Interceptor,
    transport::{Channel, Endpoint},
    Request, Status,
};

pub mod hello_world {
    tonic::include_proto!("helloworld");
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let channel = Endpoint::from_static("http://[::1]:50051")
        .connect()
        .await?;

    let mut client = GreeterClient::with_interceptor(channel, intercept);

    let request = tonic::Request::new(HelloRequest {
        name: "Tonic".into(),
    });

    let response = client.say_hello(request).await?;

    println!("RESPONSE={:?}", response);

    Ok(())
}

/// This function will get called on each outbound request. Returning a
/// `Status` here will cancel the request and have that status returned to
/// the caller.
fn intercept(req: Request<()>) -> Result<Request<()>, Status> {
    println!("Intercepting request: {:?}", req);
    Ok(req)
}

// You can also use the `Interceptor` trait to create an interceptor type
// that is easy to name
struct MyInterceptor;

impl Interceptor for MyInterceptor {
    fn call(&mut self, request: tonic::Request<()>) -> Result<tonic::Request<()>, Status> {
        Ok(request)
    }
}

#[allow(dead_code, unused_variables)]
async fn using_named_interceptor() -> Result<(), Box<dyn std::error::Error>> {
    let channel = Endpoint::from_static("http://[::1]:50051")
        .connect()
        .await?;

    let client: GreeterClient<InterceptedService<Channel, MyInterceptor>> =
        GreeterClient::with_interceptor(channel, MyInterceptor);

    Ok(())
}

// Using a function pointer type might also be possible if your interceptor is a
// bare function that doesn't capture any variables
#[allow(dead_code, unused_variables, clippy::type_complexity)]
async fn using_function_pointer_interceptro() -> Result<(), Box<dyn std::error::Error>> {
    let channel = Endpoint::from_static("http://[::1]:50051")
        .connect()
        .await?;

    let client: GreeterClient<
        InterceptedService<Channel, fn(tonic::Request<()>) -> Result<tonic::Request<()>, Status>>,
    > = GreeterClient::with_interceptor(channel, intercept);

    Ok(())
}

Notice that if your interceptor carries no state, you can also just use function pointers, see the intercept function. This would be the preferable option if your usecase is simple.

Both the interceptors showcased above are client-side interceptors, here is how server-interceptors look like:

use tonic::{transport::Server, Request, Response, Status};

use hello_world::greeter_server::{Greeter, GreeterServer};
use hello_world::{HelloReply, HelloRequest};

pub mod hello_world {
    tonic::include_proto!("helloworld");
}

#[derive(Default)]
pub struct MyGreeter {}

#[tonic::async_trait]
impl Greeter for MyGreeter {
    async fn say_hello(
        &self,
        request: Request<HelloRequest>,
    ) -> Result<Response<HelloReply>, Status> {
        let extension = request.extensions().get::<MyExtension>().unwrap();
        println!("extension data = {}", extension.some_piece_of_data);

        let reply = hello_world::HelloReply {
            message: format!("Hello {}!", request.into_inner().name),
        };
        Ok(Response::new(reply))
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let addr = "[::1]:50051".parse().unwrap();
    let greeter = MyGreeter::default();

    // See examples/src/interceptor/client.rs for an example of how to create a
    // named interceptor that can be returned from functions or stored in
    // structs.
    let svc = GreeterServer::with_interceptor(greeter, intercept);

    println!("GreeterServer listening on {}", addr);

    Server::builder().add_service(svc).serve(addr).await?;

    Ok(())
}

/// This function will get called on each inbound request, if a `Status`
/// is returned, it will cancel the request and return that status to the
/// client.
fn intercept(mut req: Request<()>) -> Result<Request<()>, Status> {
    println!("Intercepting request: {:?}", req);

    // Set an extension that can be retrieved by `say_hello`
    req.extensions_mut().insert(MyExtension {
        some_piece_of_data: "foo".to_string(),
    });

    Ok(req)
}

struct MyExtension {
    some_piece_of_data: String,
}

You can see that implementation wise, there is not much difference between client-side and server-side.

To conclude, use interceptors in tonic for simple things, for complex things use middleware. Interceptors are a general gRPC concept and so you will find them in every implementation of the protocol, whereas middlewares are a general concept unrelated to gRPC and so you need to use a 3rd party library that proxies your calls so you can insert them.

The main benefit of these is reducing code duplication that you might have been otherwise prone to create.

unary vs stream

Some implementations of gRPC may also differentiate between unary and stream

On the client-side, an interceptor may for example attach a JWT to the metadata, to facilitate said authentication, which will allow us to remove it from the protobuf message definition and optionally allow for multiple authentication methods without polluting the protocol buffers.interceptors. Clue is in the name, stream interceptors work on stream gRPC calls, whereas unary interceptors are concerned with unary calls.

When an interceptor is triggered for a stream, the trigger is executed at its creation.

Braiins University

Interceptors

Incoming and outgoing

In Rust - Tonic specifics

unary vs stream