[ETCM-48] Add json rpc http healthcheck (#713)

mirkoAlic · web-flow · commit 7cb9800cecc1 · 2020-10-01T09:54:02.000-03:00
diff --git a/insomnia_workspace.json b/insomnia_workspace.json
@@ -1,7 +1,7 @@
 {
   "_type": "export",
   "__export_format": 4,
-  "__export_date": "2020-09-28T17:05:07.127Z",
+  "__export_date": "2020-09-30T20:01:40.792Z",
   "__export_source": "insomnia.desktop.app:v2020.4.1",
   "resources": [
     {
@@ -272,6 +272,35 @@
       "metaSortKey": -1553869483792,
       "_type": "request_group"
     },
+    {
+      "_id": "req_8c4ccaa3552544a4b61bc33cc9fa547c",
+      "parentId": "fld_f75b249a780c4b5e97a0a2980ad1d4b8",
+      "modified": 1580405661758,
+      "created": 1580405461883,
+      "url": "{{ node_url }}/healthcheck",
+      "name": "healthcheck node",
+      "description": "",
+      "method": "GET",
+      "body": {},
+      "parameters": [],
+      "headers": [
+        {
+          "id": "pair_088edc31f5e04f20a16b465a673871bb",
+          "name": "Cache-Control",
+          "value": "no-cache"
+        }
+      ],
+      "authentication": {},
+      "metaSortKey": -1552939150156.3438,
+      "isPrivate": false,
+      "settingStoreCookies": true,
+      "settingSendCookies": true,
+      "settingDisableRenderRequestBody": false,
+      "settingEncodeUrl": true,
+      "settingRebuildPath": true,
+      "settingFollowRedirects": "global",
+      "_type": "request"
+    },
     {
       "_id": "req_b60c1a4f9d604d868910f967c6a070d7",
       "parentId": "fld_a06eb77e183c4727800eb7dc43ceabe1",
diff --git a/src/main/scala/io/iohk/ethereum/healthcheck/Healthcheck.scala b/src/main/scala/io/iohk/ethereum/healthcheck/Healthcheck.scala
@@ -0,0 +1,37 @@
+package io.iohk.ethereum.healthcheck
+
+import scala.concurrent.{ExecutionContext, Future}
+import scala.util.{Failure, Success}
+
+/**
+  * Represents a health check, runs it and interprets the outcome.
+  * The outcome can be either a normal result, an application error, or
+  * an (unexpected) exception.
+  *
+  * @param description An one-word description of the health check.
+  * @param f           The function that runs the health check.
+  * @param mapResultToError A function that interprets the result.
+  * @param mapErrorToError  A function that interprets the application error.
+  * @param mapExceptionToError A function that interprets the (unexpected) exception.
+  * @tparam Error  The type of the application error.
+  * @tparam Result The type of the actual value expected by normal termination of `f`.
+  */
+case class Healthcheck[Error, Result](
+    description: String,
+    f: () ⇒ Future[Either[Error, Result]],
+    mapResultToError: Result ⇒ Option[String] = (_: Result) ⇒ None,
+    mapErrorToError: Error ⇒ Option[String] = (error: Error) ⇒ Some(String.valueOf(error)),
+    mapExceptionToError: Throwable ⇒ Option[String] = (t: Throwable) ⇒ Some(String.valueOf(t))
+) {
+
+  def apply()(implicit ec: ExecutionContext): Future[HealthcheckResult] = {
+    f().transform {
+      case Success(Left(error)) ⇒
+        Success(HealthcheckResult(description, mapErrorToError(error)))
+      case Success(Right(result)) ⇒
+        Success(HealthcheckResult(description, mapResultToError(result)))
+      case Failure(t) ⇒
+        Success(HealthcheckResult(description, mapExceptionToError(t)))
+    }
+  }
+}
diff --git a/src/main/scala/io/iohk/ethereum/healthcheck/HealthcheckResponse.scala b/src/main/scala/io/iohk/ethereum/healthcheck/HealthcheckResponse.scala
@@ -0,0 +1,5 @@
+package io.iohk.ethereum.healthcheck
+
+final case class HealthcheckResponse(checks: List[HealthcheckResult]) {
+  lazy val isOK: Boolean = checks.forall(_.isOK)
+}
diff --git a/src/main/scala/io/iohk/ethereum/healthcheck/HealthcheckResult.scala b/src/main/scala/io/iohk/ethereum/healthcheck/HealthcheckResult.scala
@@ -0,0 +1,18 @@
+package io.iohk.ethereum.healthcheck
+
+final case class HealthcheckResult private (description: String, status: String, error: Option[String]) {
+  assert(
+    status == HealthcheckStatus.OK && error.isEmpty || status == HealthcheckStatus.ERROR && error.isDefined
+  )
+
+  def isOK: Boolean = status == HealthcheckStatus.OK
+}
+
+object HealthcheckResult {
+  def apply(description: String, error: Option[String]): HealthcheckResult =
+    new HealthcheckResult(
+      description = description,
+      status = error.fold(HealthcheckStatus.OK)(_ ⇒ HealthcheckStatus.ERROR),
+      error = error
+    )
+}
diff --git a/src/main/scala/io/iohk/ethereum/healthcheck/HealthcheckStatus.scala b/src/main/scala/io/iohk/ethereum/healthcheck/HealthcheckStatus.scala
@@ -0,0 +1,6 @@
+package io.iohk.ethereum.healthcheck
+
+object HealthcheckStatus {
+  final val OK = "OK"
+  final val ERROR = "ERROR"
+}
diff --git a/src/main/scala/io/iohk/ethereum/jsonrpc/JsonRpcControllerMetrics.scala b/src/main/scala/io/iohk/ethereum/jsonrpc/JsonRpcControllerMetrics.scala
@@ -13,4 +13,6 @@ case object JsonRpcControllerMetrics extends MetricsContainer {
   final val MethodsSuccessCounter = metrics.counter("json.rpc.methods.success.counter")
   final val MethodsExceptionCounter = metrics.counter("json.rpc.methods.exception.counter")
   final val MethodsErrorCounter = metrics.counter("json.rpc.methods.error.counter")
+
+  final val HealhcheckErrorCounter = metrics.counter("json.rpc.healthcheck.error.counter")
 }
diff --git a/src/main/scala/io/iohk/ethereum/jsonrpc/JsonRpcHealthChecker.scala b/src/main/scala/io/iohk/ethereum/jsonrpc/JsonRpcHealthChecker.scala
@@ -0,0 +1,21 @@
+package io.iohk.ethereum.jsonrpc
+
+import io.iohk.ethereum.healthcheck.HealthcheckResponse
+
+import scala.concurrent.Future
+import scala.util.{Failure, Success}
+import scala.concurrent.ExecutionContext.Implicits.global
+
+trait JsonRpcHealthChecker {
+  def healthCheck(): Future[HealthcheckResponse]
+
+  def handleResponse(responseF: Future[HealthcheckResponse]): Future[HealthcheckResponse] = {
+    responseF.andThen {
+      case Success(response) if (!response.isOK) =>
+        JsonRpcControllerMetrics.HealhcheckErrorCounter.increment()
+      case Failure(t) =>
+        JsonRpcControllerMetrics.HealhcheckErrorCounter.increment()
+    }
+  }
+
+}
diff --git a/src/main/scala/io/iohk/ethereum/jsonrpc/JsonRpcHealthcheck.scala b/src/main/scala/io/iohk/ethereum/jsonrpc/JsonRpcHealthcheck.scala
@@ -0,0 +1,9 @@
+package io.iohk.ethereum.jsonrpc
+
+import io.iohk.ethereum.healthcheck.Healthcheck
+
+object JsonRpcHealthcheck {
+  type T[R] = Healthcheck[JsonRpcError, R]
+
+  def apply[R](description: String, f: () ⇒ ServiceResponse[R]): T[R] = Healthcheck(description, f)
+}
diff --git a/src/main/scala/io/iohk/ethereum/jsonrpc/NodeJsonRpcHealthChecker.scala b/src/main/scala/io/iohk/ethereum/jsonrpc/NodeJsonRpcHealthChecker.scala
@@ -0,0 +1,44 @@
+package io.iohk.ethereum.jsonrpc
+
+import io.iohk.ethereum.healthcheck.HealthcheckResponse
+import io.iohk.ethereum.jsonrpc.EthService._
+import io.iohk.ethereum.jsonrpc.NetService._
+
+import scala.concurrent.ExecutionContext.Implicits.global
+import scala.concurrent.Future
+
+class NodeJsonRpcHealthChecker(
+    netService: NetService,
+    ethService: EthService
+) extends JsonRpcHealthChecker {
+
+  protected def mainService: String = "node health"
+
+  final val listeningHC = JsonRpcHealthcheck("listening", () ⇒ netService.listening(NetService.ListeningRequest()))
+  final val peerCountHC = JsonRpcHealthcheck("peerCount", () ⇒ netService.peerCount(PeerCountRequest()))
+  final val earliestBlockHC = JsonRpcHealthcheck(
+    "earliestBlock",
+    () ⇒ ethService.getBlockByNumber(BlockByNumberRequest(BlockParam.Earliest, true))
+  )
+  final val latestBlockHC = JsonRpcHealthcheck(
+    "latestBlock",
+    () ⇒ ethService.getBlockByNumber(BlockByNumberRequest(BlockParam.Latest, true))
+  )
+  final val pendingBlockHC = JsonRpcHealthcheck(
+    "pendingBlock",
+    () ⇒ ethService.getBlockByNumber(BlockByNumberRequest(BlockParam.Pending, true))
+  )
+
+  override def healthCheck(): Future[HealthcheckResponse] = {
+    val listeningF = listeningHC()
+    val peerCountF = peerCountHC()
+    val earliestBlockF = earliestBlockHC()
+    val latestBlockF = latestBlockHC()
+    val pendingBlockF = pendingBlockHC()
+
+    val allChecksF = List(listeningF, peerCountF, earliestBlockF, latestBlockF, pendingBlockF)
+    val responseF = Future.sequence(allChecksF).map(HealthcheckResponse)
+
+    handleResponse(responseF)
+  }
+}
diff --git a/src/main/scala/io/iohk/ethereum/jsonrpc/server/http/BasicJsonRpcHttpServer.scala b/src/main/scala/io/iohk/ethereum/jsonrpc/server/http/BasicJsonRpcHttpServer.scala
@@ -9,9 +9,13 @@ import io.iohk.ethereum.utils.Logger
 import scala.concurrent.ExecutionContext.Implicits.global
 import scala.util.{Failure, Success}
 
-class BasicJsonRpcHttpServer(val jsonRpcController: JsonRpcController, config: JsonRpcHttpServerConfig)
-                            (implicit val actorSystem: ActorSystem)
-  extends JsonRpcHttpServer with Logger {
+class BasicJsonRpcHttpServer(
+    val jsonRpcController: JsonRpcController,
+    val jsonRpcHealthChecker: JsonRpcHealthChecker,
+    config: JsonRpcHttpServerConfig
+)(implicit val actorSystem: ActorSystem)
+    extends JsonRpcHttpServer
+    with Logger {
 
   def run(): Unit = {
     val bindingResultF = Http(actorSystem).newServerAt(config.interface, config.port).bind(route)
diff --git a/src/main/scala/io/iohk/ethereum/jsonrpc/server/http/JsonRpcHttpServer.scala b/src/main/scala/io/iohk/ethereum/jsonrpc/server/http/JsonRpcHttpServer.scala
@@ -1,15 +1,15 @@
 package io.iohk.ethereum.jsonrpc.server.http
 
 import akka.actor.ActorSystem
-import akka.http.scaladsl.model.StatusCodes
+import akka.http.scaladsl.model._
 import akka.http.scaladsl.server.Directives._
-import akka.http.scaladsl.server.{MalformedRequestContentRejection, RejectionHandler, Route}
+import akka.http.scaladsl.server._
 import ch.megard.akka.http.cors.javadsl.CorsRejection
 import ch.megard.akka.http.cors.scaladsl.CorsDirectives._
 import ch.megard.akka.http.cors.scaladsl.model.HttpOriginMatcher
 import ch.megard.akka.http.cors.scaladsl.settings.CorsSettings
 import de.heikoseeberger.akkahttpjson4s.Json4sSupport
-import io.iohk.ethereum.jsonrpc.{JsonRpcController, JsonRpcErrors, JsonRpcRequest, JsonRpcResponse}
+import io.iohk.ethereum.jsonrpc._
 import io.iohk.ethereum.utils.{ConfigUtils, Logger}
 import java.security.SecureRandom
 import org.json4s.JsonAST.JInt
@@ -20,6 +20,7 @@ import scala.util.Try
 
 trait JsonRpcHttpServer extends Json4sSupport {
   val jsonRpcController: JsonRpcController
+  val jsonRpcHealthChecker: JsonRpcHealthChecker
 
   implicit val serialization = native.Serialization
 
@@ -32,7 +33,8 @@ trait JsonRpcHttpServer extends Json4sSupport {
     .withAllowedOrigins(corsAllowedOrigins)
 
   implicit def myRejectionHandler: RejectionHandler =
-    RejectionHandler.newBuilder()
+    RejectionHandler
+      .newBuilder()
       .handle {
         case _: MalformedRequestContentRejection =>
           complete((StatusCodes.BadRequest, JsonRpcResponse("2.0", None, Some(JsonRpcErrors.ParseError), JInt(0))))
@@ -42,7 +44,9 @@ trait JsonRpcHttpServer extends Json4sSupport {
       .result()
 
   val route: Route = cors(corsSettings) {
-    (pathEndOrSingleSlash & post) {
+    (path("healthcheck") & pathEndOrSingleSlash & get) {
+      handleHealthcheck()
+    } ~ (pathEndOrSingleSlash & post) {
       entity(as[JsonRpcRequest]) { request =>
         handleRequest(request)
       } ~ entity(as[Seq[JsonRpcRequest]]) { request =>
@@ -56,6 +60,24 @@ trait JsonRpcHttpServer extends Json4sSupport {
     */
   def run(): Unit
 
+  private def handleHealthcheck(): StandardRoute = {
+    val responseF = jsonRpcHealthChecker.healthCheck()
+    val httpResponseF =
+      responseF.map {
+        case response if response.isOK ⇒
+          HttpResponse(
+            status = StatusCodes.OK,
+            entity = HttpEntity(ContentTypes.`application/json`, serialization.writePretty(response))
+          )
+        case response ⇒
+          HttpResponse(
+            status = StatusCodes.InternalServerError,
+            entity = HttpEntity(ContentTypes.`application/json`, serialization.writePretty(response))
+          )
+      }
+    complete(httpResponseF)
+  }
+
   private def handleRequest(request: JsonRpcRequest) = {
     complete(jsonRpcController.handleRequest(request))
   }
@@ -67,12 +89,18 @@ trait JsonRpcHttpServer extends Json4sSupport {
 
 object JsonRpcHttpServer extends Logger {
 
-  def apply(jsonRpcController: JsonRpcController, config: JsonRpcHttpServerConfig, secureRandom: SecureRandom)
-           (implicit actorSystem: ActorSystem): Either[String, JsonRpcHttpServer] = config.mode match {
-    case "http" => Right(new BasicJsonRpcHttpServer(jsonRpcController, config)(actorSystem))
-    case "https" => Right(new JsonRpcHttpsServer(jsonRpcController, config, secureRandom)(actorSystem))
-    case _ => Left(s"Cannot start JSON RPC server: Invalid mode ${config.mode} selected")
-  }
+  def apply(
+      jsonRpcController: JsonRpcController,
+      jsonRpcHealthchecker: JsonRpcHealthChecker,
+      config: JsonRpcHttpServerConfig,
+      secureRandom: SecureRandom
+  )(implicit actorSystem: ActorSystem): Either[String, JsonRpcHttpServer] =
+    config.mode match {
+      case "http" => Right(new BasicJsonRpcHttpServer(jsonRpcController, jsonRpcHealthchecker, config)(actorSystem))
+      case "https" =>
+        Right(new JsonRpcHttpsServer(jsonRpcController, jsonRpcHealthchecker, config, secureRandom)(actorSystem))
+      case _ => Left(s"Cannot start JSON RPC server: Invalid mode ${config.mode} selected")
+    }
 
   trait JsonRpcHttpServerConfig {
     val mode: String
@@ -99,9 +127,15 @@ object JsonRpcHttpServer extends Logger {
 
         override val corsAllowedOrigins = ConfigUtils.parseCorsAllowedOrigins(rpcHttpConfig, "cors-allowed-origins")
 
-        override val certificateKeyStorePath: Option[String] = Try(rpcHttpConfig.getString("certificate-keystore-path")).toOption
-        override val certificateKeyStoreType: Option[String] = Try(rpcHttpConfig.getString("certificate-keystore-type")).toOption
-        override val certificatePasswordFile: Option[String] = Try(rpcHttpConfig.getString("certificate-password-file")).toOption
+        override val certificateKeyStorePath: Option[String] = Try(
+          rpcHttpConfig.getString("certificate-keystore-path")
+        ).toOption
+        override val certificateKeyStoreType: Option[String] = Try(
+          rpcHttpConfig.getString("certificate-keystore-type")
+        ).toOption
+        override val certificatePasswordFile: Option[String] = Try(
+          rpcHttpConfig.getString("certificate-password-file")
+        ).toOption
       }
     }
   }
diff --git a/src/main/scala/io/iohk/ethereum/jsonrpc/server/http/JsonRpcHttpsServer.scala b/src/main/scala/io/iohk/ethereum/jsonrpc/server/http/JsonRpcHttpsServer.scala
diff --git a/src/main/scala/io/iohk/ethereum/nodebuilder/NodeBuilder.scala b/src/main/scala/io/iohk/ethereum/nodebuilder/NodeBuilder.scala
diff --git a/src/test/scala/io/iohk/ethereum/jsonrpc/server/http/JsonRpcHttpServerSpec.scala b/src/test/scala/io/iohk/ethereum/jsonrpc/server/http/JsonRpcHttpServerSpec.scala

Original file line number	Diff line number	Diff line change
`@@ -13,4 +13,6 @@ case object JsonRpcControllerMetrics extends MetricsContainer {`
`13`	`13`	`final val MethodsSuccessCounter = metrics.counter("json.rpc.methods.success.counter")`
`14`	`14`	`final val MethodsExceptionCounter = metrics.counter("json.rpc.methods.exception.counter")`
`15`	`15`	`final val MethodsErrorCounter = metrics.counter("json.rpc.methods.error.counter")`
	`16`	`+`
	`17`	`+ final val HealhcheckErrorCounter = metrics.counter("json.rpc.healthcheck.error.counter")`
`16`	`18`	`}`