gRPC ServerのExceptionFilterの方法をまとめた（grpc-java）

August 18, 2017
gRPC
Exception
Filter
Interceptor

前回のエントリでは認証処理やメトリクス計測、ログ出力などをインターセプターをつかい横断的に処理する方法をまとめた。

Kotlin + gRPCでio.grpc.Contextをつかってログ出力を横断処理してみた - 平日インプット週末アウトプットぶろぐ

今回のエントリではgRPC Serverのトランザクション内で発生した例外処理を横断的にキャッチしてレスポンスを返す方法をまとめていきたい。

try-catchして例外クラスに応じた処理を施す

愚直に書けば次のようになるだろう。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21


override fun getTaskService(request: TaskInbound?, responseObserver: StreamObserver&lt;TaskOutbound>?) {
    try {
        val taskId = GRpcInboundValidator.validTaskInbound(request)

        val log = GRpcLogContextHandler.getLog()
        log.elem { "taskId" to taskId }

        val task = getTaskService(GetTaskCommand(taskId.toLong()))
        val msg = getOutbound(task)
        responseObserver?.onNext(msg)
        responseObserver?.onCompleted()
    } catch (e: WebAppException.NotFoundException) {
        logger.error { "gRPC server error, task not found." }
        responseObserver?.onError(
                Status.NOT_FOUND.withDescription("task not found.").asRuntimeException())
    } catch (e: WebAppException.BadRequestException) {
        logger.error { "gRPC server error, invalid request." }
        responseObserver?.onError(
                Status.INVALID_ARGUMENT.withDescription("invalid request.").asRuntimeException())
    }
}

タスクを１件取得するgRPC Serverのコード例である。
タスクがなかった場合にはNOT FOUNDのレスポンス、リクエストに不備があればINVALID_ARGUMENTのレスポンスを返していることがわかる。
レスポンスを返す直前にはExceptionをキャッチしている。そしてExceptionごとにエラーレスポンスを並べている形だ。

このような書き方を複数のgRPC Serverで続けるとcatch節が冗長になってしまう。
gRPC ServerのインターセプターをつかいExceptionのキャッチを共通処理としていきたい。そうすれば上記のコードからtry-catchが無くなりコードの見通しが良くなる。

インターセプターをつくる

gRPC Serverのトランザクションで発生した例外をキャッチするには次のようなインターセプターを用意する。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29


@Component
class ExceptionFilter : ServerInterceptor {

    private val logger = KotlinLogging.logger {}

    override fun &lt;ReqT : Any?, RespT : Any?> interceptCall(call: ServerCall&lt;ReqT, RespT>?, headers: Metadata?,
                                                           next: ServerCallHandler&lt;ReqT, RespT>?): ServerCall.Listener&lt;ReqT> {
        return object : ForwardingServerCallListener.SimpleForwardingServerCallListener&lt;ReqT>(next?.startCall(call, headers)!!) {

            override fun onHalfClose() {
                try {
                    super.onHalfClose()
                } catch (ex: RuntimeException) {
                    handleException(call, headers, ex)
                    throw ex
                }
            }

            override fun onReady() {
                try {
                    super.onReady()
                } catch (ex: RuntimeException) {
                    handleException(call, headers, ex)
                    throw ex
                }

            }
        }
    }

onHalfCloseとonReadyをoverrideしたリスナーを用意する。onHalfCloseとonReadyはクライアントからリクエストが送信された後、つまりServer内の処理のリスナーである。
この２つリスナー関数をtry-catchで囲むことでgRPC Server内の例外エラーを横断的にキャッチすることができる。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18


private fun &lt;ReqT, RespT> handleException(call: ServerCall&lt;ReqT, RespT>?, headers: Metadata?, ex: Exception) {

    logger.error(ex) { ex.message }

    when (ex) {
        is RepositoryException.NotFoundException -> call?.close(
                Status.fromCode(Status.NOT_FOUND.code).withDescription(ex.message), headers)
        is RepositoryException.ConflictException -> call?.close(
                Status.fromCode(Status.ALREADY_EXISTS.code).withDescription(ex.message), headers)
        is WebAppException.BadRequestException -> call?.close(
                Status.fromCode(Status.INVALID_ARGUMENT.code).withDescription(ex.message), headers)
        is WebAppException.NotFoundException -> call?.close(
                Status.fromCode(Status.NOT_FOUND.code).withDescription(ex.message), headers)
        is EmptyResultDataAccessException -> call?.close(
                Status.fromCode(Status.NOT_FOUND.code).withDescription("data not found."), headers)
        else -> call?.close(Status.fromCode(Status.INTERNAL.code).withDescription(ex.message), headers)
    }
}

handleException関数では例外クラスに応じたれステータスコードを定義してcall?.closeを呼び出している。
これでExceptionFilterが例外エラーをすべてキャッチしてレスポンスを返すことができる。

gRPC Serverのテスト時にExceptionFilterをつかう

gRPC ServerでExceptionFilterクラスをテストしたい。その場合にはテスト時にInProcessServerのビルダーにinterceptを加えるとよい。

1
2
3
4
5
6


inProcessServer = InProcessServerBuilder
        .forName(UNIQUE_SERVER_NAME)
        .addService(target)
        .intercept(ExceptionFilter())  ← こちら
        .directExecutor()
        .build()

まとめ

今回の例外エラーのキャッチ方法は情報が少なくgrpc-javaのissueのコメントで少し言及されていたところからヒントを得た。
振り返るとServerCallListenerのドキュメントを読んだうえで、それぞれのリスナーの役割を理解できれば実装できるものだった。
gRPCの実用化に向けてインターセプターのネタも揃ってきた。

コード

エントリで紹介したコードは一部分のためコード全体はgithubを参照してください。

インターセプターのコードはこちらです。