airframe-rpc: Add RPC status code (#1973)

* Define standard RPC status code
* Add gRPC status code
* Add PackSupport to serialize error code
* Add HttpStatus -> GrpcStatus mapping
* Fix gRPC status mapping

Author: xerial

airframe-codec/src/main/scala/wvlet/airframe/codec/PackSupport.scala

@@ -0,0 +1,27 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package wvlet.airframe.codec
+
+import wvlet.airframe.msgpack.spi.Packer
+
+/**
+  * If a class is extending PackSupport, pack(Packer) method will be used for serializing (packing) this object.
+  *
+  * For deserializing (unpacking) data of type A, define a package method unapply(string:String): Option[A].
+  *
+  * In future, we may support unapply(u:Unpacker): Option[A] method
+  */
+trait PackSupport {
+  def pack(packer: Packer): Unit
+}

airframe-codec/src/main/scala/wvlet/airframe/codec/PrimitiveCodec.scala

@@ -921,19 +921,20 @@ object PrimitiveCodec {
       v match {
         case null => p.packNil
         // Primitive types
-        case v: String    => StringCodec.pack(p, v)
-        case v: Boolean   => BooleanCodec.pack(p, v)
-        case v: Int       => IntCodec.pack(p, v)
-        case v: Long      => LongCodec.pack(p, v)
-        case v: Float     => FloatCodec.pack(p, v)
-        case v: Double    => DoubleCodec.pack(p, v)
-        case v: Byte      => ByteCodec.pack(p, v)
-        case v: Short     => ShortCodec.pack(p, v)
-        case v: Char      => CharCodec.pack(p, v)
-        case v: JSONValue => JSONValueCodec.pack(p, v)
-        case v: Value     => ValueCodec.pack(p, v)
-        case v: Instant   => p.packTimestamp(v)
-        case v: ULID      => ULIDCodec.pack(p, v)
+        case v: String       => StringCodec.pack(p, v)
+        case v: Boolean      => BooleanCodec.pack(p, v)
+        case v: Int          => IntCodec.pack(p, v)
+        case v: Long         => LongCodec.pack(p, v)
+        case v: Float        => FloatCodec.pack(p, v)
+        case v: Double       => DoubleCodec.pack(p, v)
+        case v: Byte         => ByteCodec.pack(p, v)
+        case v: Short        => ShortCodec.pack(p, v)
+        case v: Char         => CharCodec.pack(p, v)
+        case v: JSONValue    => JSONValueCodec.pack(p, v)
+        case v: Value        => ValueCodec.pack(p, v)
+        case v: Instant      => p.packTimestamp(v)
+        case ps: PackSupport => ps.pack(p)
+        case v: ULID         => ULIDCodec.pack(p, v)
         // Arrays
         case v: Array[String]  => StringArrayCodec.pack(p, v)
         case v: Array[Boolean] => BooleanArrayCodec.pack(p, v)

airframe-config/src/main/scala/wvlet/airframe/config/YamlReader.scala

@@ -148,7 +148,7 @@ class YamlReader(map: Map[AnyRef, AnyRef]) extends LogSupport {
             pack(packer, k)
             pack(packer, v)
           }
-        case c if ReflectTypeUtil.isJavaColleciton(c.getClass) =>
+        case c if ReflectTypeUtil.isJavaCollection(c.getClass) =>
           val cl = c.asInstanceOf[java.util.Collection[_]].asScala
           trace(s"pack collection (${cl.size})")
           packer.packArrayHeader(cl.size)

airframe-http-grpc/src/main/scala/wvlet/airframe/http/grpc/internal/GrpcException.scala

@@ -13,9 +13,9 @@
  */
 package wvlet.airframe.http.grpc.internal
 
-import io.grpc.{StatusException, StatusRuntimeException}
+import io.grpc.{Metadata, Status, StatusException, StatusRuntimeException}
 import wvlet.airframe.codec.MessageCodecException
-import wvlet.airframe.http.{HttpServerException, HttpStatus}
+import wvlet.airframe.http.{GrpcStatus, HttpServerException, HttpStatus}
 import wvlet.log.LogSupport
 
 import java.lang.reflect.InvocationTargetException
@@ -26,6 +26,8 @@ import scala.concurrent.ExecutionException
   */
 object GrpcException extends LogSupport {
 
+  private[grpc] val rpcErrorKey = Metadata.Key.of[String]("airframe_rpc_error", Metadata.ASCII_STRING_MARSHALLER)
+
   /**
     * Convert an exception to gRPC-specific exception types
     *
@@ -56,40 +58,34 @@ object GrpcException extends LogSupport {
         s
       case e: MessageCodecException =>
         io.grpc.Status.INTERNAL
-          .withDescription(s"Failed to encode/decode data")
+          .withDescription(s"Failed to encode/decode data: ${e.getMessage}")
           .withCause(e)
           .asRuntimeException()
       case e: IllegalArgumentException =>
         io.grpc.Status.INVALID_ARGUMENT
           .withCause(e)
+          .withDescription(e.getMessage)
           .asRuntimeException()
       case e: UnsupportedOperationException =>
         io.grpc.Status.UNIMPLEMENTED
           .withCause(e)
+          .withDescription(e.getMessage)
           .asRuntimeException()
       case e: HttpServerException =>
-        val s = e.status match {
-          case HttpStatus.BadRequest_400 =>
-            io.grpc.Status.INVALID_ARGUMENT
-          case HttpStatus.Unauthorized_401 =>
-            io.grpc.Status.UNAUTHENTICATED
-          case HttpStatus.Forbidden_403 =>
-            io.grpc.Status.PERMISSION_DENIED
-          case HttpStatus.NotFound_404 =>
-            io.grpc.Status.UNIMPLEMENTED
-          case HttpStatus.Conflict_409 =>
-            io.grpc.Status.ALREADY_EXISTS
-          case HttpStatus.TooManyRequests_429 | HttpStatus.BadGateway_502 | HttpStatus.ServiceUnavailable_503 |
-              HttpStatus.GatewayTimeout_504 =>
-            io.grpc.Status.UNAVAILABLE
-          case s if s.isServerError =>
-            io.grpc.Status.INTERNAL
-          case s if s.isClientError =>
-            io.grpc.Status.INVALID_ARGUMENT
-          case other =>
-            io.grpc.Status.UNKNOWN
+        val grpcStatus = GrpcStatus.ofHttpStatus(e.status)
+        val s = Status
+          .fromCodeValue(grpcStatus.code)
+          .withCause(e)
+          .withDescription(e.getMessage)
+
+        if (e.message.nonEmpty) {
+          val m        = e.message
+          val metadata = new Metadata()
+          metadata.put[String](rpcErrorKey, s"${m.toContentString}")
+          s.asRuntimeException(metadata)
+        } else {
+          s.asRuntimeException()
         }
-        s.withCause(e).asRuntimeException()
       case other =>
         io.grpc.Status.INTERNAL
           .withCause(other)

airframe-http-grpc/src/main/scala/wvlet/airframe/http/grpc/internal/GrpcRequestHandler.scala

@@ -302,7 +302,9 @@ class GrpcRequestHandler(
 
 }
 
-private[grpc] class RPCUnaryMethodHandler(rpcRequestHandler: GrpcRequestHandler) extends UnaryMethod[MsgPack, Any] {
+private[grpc] class RPCUnaryMethodHandler(rpcRequestHandler: GrpcRequestHandler)
+    extends UnaryMethod[MsgPack, Any]
+    with LogSupport {
   override def invoke(
       request: MsgPack,
       responseObserver: StreamObserver[Any]

airframe-http-grpc/src/main/scala/wvlet/airframe/http/grpc/internal/GrpcResponseHeaderInterceptor.scala

@@ -43,6 +43,6 @@ private[grpc] object GrpcResponseHeaderInterceptor extends ServerInterceptor {
         }
       },
       requestHeaders
-    );
+    )
   }
 }

airframe-http-grpc/src/test/scala/wvlet/airframe/http/grpc/GrpcErrorHandlingTest.scala

@@ -0,0 +1,46 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package wvlet.airframe.http.grpc
+
+import io.grpc.Status
+import io.grpc.Status.Code
+import io.grpc.StatusRuntimeException
+import wvlet.airframe.http.Router
+import wvlet.airframe.http.grpc.GrpcErrorLogTest.DemoApiDebug
+import wvlet.airframe.http.grpc.example.DemoApi.DemoApiClient
+import wvlet.airframe.http.grpc.internal.GrpcException
+import wvlet.airspec.AirSpec
+
+object GrpcErrorHandlingTest extends AirSpec {
+
+  protected override def design = {
+    gRPC.server
+      .withName("error-handling-test-api")
+      .withRouter(Router.add[DemoApiDebug])
+      .designWithChannel
+  }
+
+  test("handle error") { (client: DemoApiClient) =>
+    warn("Starting a gRPC error handling test")
+    val ex = intercept[StatusRuntimeException] {
+      client.error409Test
+    }
+    ex.getMessage.contains("409") shouldBe true
+    ex.getStatus.isOk shouldBe false
+    ex.getStatus.getCode shouldBe Code.ABORTED
+    val trailers = Status.trailersFromThrowable(ex)
+    val rpcError = trailers.get[String](GrpcException.rpcErrorKey)
+    info(s"error trailer: ${rpcError}")
+  }
+}

airframe-http-grpc/src/test/scala/wvlet/airframe/http/grpc/example/DemoApi.scala

@@ -20,7 +20,7 @@ import wvlet.airframe.codec.MessageCodecFactory
 import wvlet.airframe.http.grpc.internal.GrpcServiceBuilder
 import wvlet.airframe.http.grpc._
 import wvlet.airframe.http.router.Route
-import wvlet.airframe.http.{RPC, Router}
+import wvlet.airframe.http.{Http, HttpStatus, RPC, Router}
 import wvlet.airframe.msgpack.spi.MsgPack
 import wvlet.airframe.rx.{Rx, RxStream}
 import wvlet.log.LogSupport
@@ -61,6 +61,10 @@ trait DemoApi extends LogSupport {
     // do nothing
     debug(s"hello ${name}")
   }
+
+  def error409Test: String = {
+    throw Http.serverException(HttpStatus.Conflict_409).withContent("test message")
+  }
 }
 
 object DemoApi {
@@ -79,6 +83,13 @@ object DemoApi {
     }
   }
 
+  /**
+    * Manually build a gRPC client here as we can't use sbt-airframe.
+    * @param channel
+    * @param callOptions
+    * @param codecFactory
+    * @param encoding
+    */
   case class DemoApiClient(
       channel: Channel,
       callOptions: CallOptions = CallOptions.DEFAULT,
@@ -105,6 +116,8 @@ object DemoApi {
       GrpcServiceBuilder.buildMethodDescriptor(getRoute("helloOpt"), codecFactory)
     private val returnUnitMethodDescriptor =
       GrpcServiceBuilder.buildMethodDescriptor(getRoute("returnUnit"), codecFactory)
+    private val errorTestMethodDescriptor =
+      GrpcServiceBuilder.buildMethodDescriptor(getRoute("error409Test"), codecFactory)
 
     def withEncoding(encoding: GrpcEncoding): DemoApiClient = {
       this.copy(encoding = encoding)
@@ -189,6 +202,13 @@ object DemoApi {
         .blockingUnaryCall(_channel, returnUnitMethodDescriptor, getCallOptions, encode(m))
       resp.asInstanceOf[Unit]
     }
+
+    def error409Test: String = {
+      val resp = ClientCalls
+        .blockingUnaryCall(_channel, errorTestMethodDescriptor, getCallOptions, encode(Map.empty))
+
+      resp.asInstanceOf[String]
+    }
   }
 
 }