gRPC インターセプターの利用（アイデア編）

このドキュメントの内容

gRPC のインターセプターの利用アイデアを考えてみます。インターセプターの仕組み（C#実装）についても簡単に説明します。

インターセプターの仕組み

インターセプターはRPCメソッドの呼び出しに対する割り込みを行うための機能です。RPCメソッドの呼び出し前／後に任意の処理を実行したり、RPCメソッドの処理を置き換えたりするときに使用します。プログラム言語によって実装は異なりますが、概念は同じようです。

gRPC.Core の C# 実装の場合、クライアントサイド用に CallInvoker クラスと Channel クラス、サーバーサイド用に ServerServiceDefinition クラスに対する拡張メソッドが追加されています。

【GitHub】Grpc.Core.Interceptors.Interceptor クラス（C#）

なお、Grpc.Auth で実装されている AsyncAuthInterceptor はデリゲートで、ここで説明するインターセプターとは全く関係はありません。

クライアントサイド

CallInvokerインスタンスまたはChannelインスタンスに対してインターセプターを設定します。
次の例では A/B/C の三つのインターセプターを設定しています。この場合、RPCメソッドが呼び出されると Aの割込処理 → Bの割込処理 → Cの割込処理 → RPCメソッドの呼び出しの順に実行されます。

CallInvoker callInvoker = new DefaultCallInvoker(channel)
    .Intercept(new IntercepterC())
    .Intercept(new IntercepterB())
    .Intercept(new IntercepterA());

SampleClient client = new SampleClient(callInvoker);

// Channel.Intercept 拡張メソッドは内部で DefaultCallInvoker を生成して返しています。
CallInvoker callInvoker = new Channel("localhost:50000", ChannelCredentials.Insecure)
    .Intercept(new IntercepterC())
    .Intercept(new IntercepterB())
    .Intercept(new IntercepterA());

SampleClient client = new SampleClient(callInvoker);

サーバーサイド

ServerServiceDefinitionインスタンスに対してインターセプターを設定します。
前述のクライアントサイドの例と同様、A/B/C の三つのインターセプターを設定しています。こちらも設定した逆順に、Aの割込処理 → Bの割込処理 → Cの割込処理 → RPCメソッドの実行の順に実行されます。

SampleService service = new SampleService();

ServerServiceDefinition definition = SampleServiceBase.BuildService(service)
    .Intercept(new InterceptorC())
    .Intercept(new InterceptorB())
    .Intercept(new InterceptorA())
;

server.Services.Add(definition);

インターセプターの実装

Grpc.Core.Interceptors.Interceptor クラスを継承して実装します。Interceptor クラスにはクライアントサイド用とサーバーサイド用の両方のメソッドが定義されていますが、必要な方だけをオーバーライドすればよいです。
RPCメソッド名やリクエストヘッダーなどの情報は context から取得できます。

以下は Interceptor クラスに定義されているメソッドの一部です。

// ClientStreaming クライアントサイド用
public virtual AsyncClientStreamingCall<TRequest, TResponse> AsyncClientStreamingCall<TRequest, TResponse>(
    ClientInterceptorContext<TRequest, TResponse> context
    , AsyncClientStreamingCallContinuation<TRequest, TResponse> continuation
){
    return continuation(request, context);
}

// ClientStreaming サーバーサイド用
public virtual Task<TResponse> ClientStreamingServerHandler<TRequest, TResponse>(
    IAsyncStreamReader<TRequest> requestStream
    , ServerCallContext context
    , ClientStreamingServerMethod<TRequest, TResponse> continuation
){
    return continuation(requestStream, context);
}

クライアントサイド用メソッドの簡単な実装例です。RPCメソッドの呼び出し前／後にログ出力を割り込ませています。

public override AsyncClientStreamingCall<TRequest, TResponse> AsyncClientStreamingCall<TRequest, TResponse>(
    ClientInterceptorContext<TRequest, TResponse> context
    , AsyncClientStreamingCallContinuation<TRequest, TResponse> continuation
){
    WriteCallingLog(context);
    Stopwatch stopwatch = Stopwatch.StartNew();
    try
    {
        return base.AsyncClientStreamingCall(context, continuation);
    }
    finally
    {
        stopwatch.Stop();
        WriteCalledLog(context, stopwatch.ElapsedMilliseconds);
    }
}

インターセプターの利用アイデア

ログやメトリクスの出力

前述の実装例のようにRPCメソッドの呼び出し前／後にログ出力を割り込ませます。

リクエストストリームに対する書き込み（WriteAsync/CompleteAsync）とレスポンスストリームに対する読み込み（MoveNext）を監視するには、インターセプターを通過するときにストリームリーダー／ライターを置き換えます。インターセプターに渡されてきたストリームリーダー／ライターを内包し、WriteAsync/CompleteAsync/MoveNext メソッドが呼び出されたときにログを出力するようにしたラッパークラスを実装します。

認証制御

RPCメソッドの呼び出し前に認証処理を行い、拒否する場合に StatusCode.Unauthenticated を返すように実装します。
IDやトークンなどの認証情報はリクエストヘッダーに格納します。

サービスダウン制御

メンテナンスやサービス時間外のRPCメソッド呼び出しを止めます。
RPCメソッドの呼び出し前に判定を行い、止める場合に StatusCode.Unavailable を返すように実装します。

例外処理

ステータスコードやエラーメッセージを置き換えます。
RPCメソッドの呼び出し時にスローされた例外をキャッチしてハンドリングします。
但し、例外をなかったことにして代理のレスポンスを返すような目的はインターセプターで対応するものではなく、個々のRPCメソッドで対応したり、アプリケーションレベルのクライアントサイド共通処理として対応したほうがよいと思います。

タイムアウト系テスト

インターセプターは、処理時間がかかったりタイムアウトが発生したときの動作テストを簡単に行うことができます。
RPCメソッドの呼び出しをディレイさせるインターセプターを用意し、必要なときのみ組み込みます。

モックを使ったテスト

インターセプターに渡されてきたストリームリーダー／ライターや continuation を置き換えれば、RPCメソッドの処理そのものを置き換えることができます。素直にモックサーバーを用意したほうがよいようにも感じますが、処理を組み込んだり外したりすることが簡単であるという点で手段の一つになりうると思います。