interceptorを使う

gRPCには最近interceptorというのが入った．厳密に言うと，各言語ごとに進捗が違っていて，goやpython，grpc-javaではinterceptorが使えるが，ruby gemとしてのgrpcにはまだinterceptorのコードが入っていない（masterには用意されている）．

interceptorとはその名の通り．イメージとしては，railsで言うところのrack middlwareや，playで言うところのfilter，goのechoでいうところのmiddlewareに近い何かだ．

RESTではよくあるけど，gRPCでコレがあるのは嬉しい．

使い方としては，io.grpc.ServerInterceptor を継承したクラスを作って，サーバ起動時に差しこめば良い．

差し込み方は，

class Logging ServerInterceptor {
//...
}

という定義をしていたら，

class GrpcServer(executionContext: ExecutionContext) { self =>
  private[this] var server: Server = null

  def start(): Unit = {
    server = ServerBuilder.forPort(55051).addService(
      ServerInterceptors.intercept(
        UsersGrpc.bindService(new UsersImpl, executionContext),
        new Logging
      )
    ).build.start
  }
  //...
}

という感じで差しこめば良い．

ログ用のinterceptor

では，ログを吐き出すinterceptorをつくろう．

class Logging() extends ServerInterceptor {

  override def interceptCall[ReqT, RespT](
    serverCall: ServerCall[ReqT, RespT],
    headers: Metadata,
    next: ServerCallHandler[ReqT, RespT]): ServerCall.Listener[ReqT] = {

    val wrapperCall: ServerCall[ReqT, RespT] = new ForwardingServerCall.SimpleForwardingServerCall[ReqT, RespT](serverCall) {
      override def request(numMessages: Int) {
        Logger.info("Request: " + headers.toString() + ", " + serverCall.getAttributes().toString() + ", " + serverCall.getMethodDescriptor().getFullMethodName())
        super.request(numMessages)
      }

      override def sendMessage(message: RespT) {
        Logger.info("Response: " + message.toString())
        super.sendMessage(message)
      }

      override def close(status: Status, trailers: Metadata) {
        Logger.info("Response: " + status.toString())
        super.close(status, trailers)
      }
    }
    return next.startCall(wrapperCall, headers)
  }
}

こんな感じで良い．

requestはこのgRPCサーバにリクエストが来た時に呼ばれる．
sendMessageはレスポンスを返すときに呼ばれる．streamを返すときは，1レスポンスごとに呼ばれる．
closeはレスポンスを全て返し終わって終了するときに呼ばれる（もしくは例外で終わるとき）．

Loggerは自分で定義するなり，好きなLoggerを使って問題ない．

これでアクセスごとにログが出るようになる．

リクエストIDをログに出したい

上記の方法でログは出力されるが，１アクセスで複数行のログが出力されているはずだ．

そうなってくると，例えば例外になったとき，それがどのrequestだったのかがわからなくなってしまう．というわけで，ログの中に一意なリクエストIDを埋めて，トレースできるようにしたい．

MDCを使う

これにはMDCを使う．

qiita.com

まず，リクエストIDを埋めたMDCContextを作る． IDをheaderから取りたい（gRPCクライアント側でIDを埋めておく）とかあると思うので，IDの決め方は好きにして欲しい．

import io.grpc.{Metadata}
import java.util.UUID
import collection.JavaConverters._
import collection.mutable._

object MDCContext {
  object Request {
    val key: String = "request_id"

    def id(): String =  {
      UUID.randomUUID().toString
    }
  }

  def buildMdcContext(headers: Metadata): java.util.Map[String, String] = {
    HashMap(Request.key -> Request.id(headers)).asJava
  }
}

注意すべきなのは，MDCContextは java.util.Map である必要がある．そのため，asJava して変換している．

そして，コレに合わせてlogback.xmlに request_id が出力されるような記述をしておく必要がある．

  <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
    <encoder>
      <pattern>%coloredLevel %logger{15} - [%X{request_id}] %message%n%xException{10}</pattern>
    </encoder>
  </appender>

こんな感じ．

MDCContextを埋める

先ほどのLoggingのinterceptorを改造する．

class Logging() extends ServerInterceptor {

  override def interceptCall[ReqT, RespT](
    serverCall: ServerCall[ReqT, RespT],
    headers: Metadata,
    next: ServerCallHandler[ReqT, RespT]): ServerCall.Listener[ReqT] = {

    // リクエストごとに固有のrequestIdを作る
    val mdcContext: java.util.Map[String, String] = MDCContext.buildMdcContext(headers)
    MDC.setContextMap(mdcContext)

    val wrapperCall: ServerCall[ReqT, RespT] = new ForwardingServerCall.SimpleForwardingServerCall[ReqT, RespT](serverCall) {
      override def request(numMessages: Int) {
        MDC.setContextMap(mdcContext)
        Logger.info("Request: " + headers.toString() + ", " + serverCall.getAttributes().toString() + ", " + serverCall.getMethodDescriptor().getFullMethodName())
        super.request(numMessages)
      }

      override def sendMessage(message: RespT) {
        MDC.setContextMap(mdcContext)
        Logger.info("Response: " + message.toString())
        super.sendMessage(message)
      }

      override def close(status: Status, trailers: Metadata) {
        MDC.setContextMap(mdcContext)
        Logger.info("Response: " + status.toString())
        super.close(status, trailers)
        MDC.clear()
      }
    }
    val listener = next.startCall(wrapperCall, headers)
    return new ForwardingServerCallListener.SimpleForwardingServerCallListener[ReqT](listener) {

      override def onMessage(message: ReqT) {
        MDC.setContextMap(mdcContext)
        super.onMessage(message)
      }

      override def onHalfClose() {
        MDC.setContextMap(mdcContext)
        super.onHalfClose()
      }

      override def onCancel() {
        MDC.setContextMap(mdcContext)
        super.onCancel()
        MDC.clear()
      }

      override def onComplete() {
        MDC.setContextMap(mdcContext)
        super.onComplete()
        MDC.clear()
      }

      override def onReady() {
        MDC.setContextMap(mdcContext)
        super.onReady()
      }
    }
  }
}